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

feat: port session 08 — Client Connection & PROXY Protocol

Browse Source 
- ClientConnection: full connection lifecycle, string/identity helpers,
  SplitSubjectQueue, KindString, MsgParts, SetHeader, message header
  manipulation (GenHeader, RemoveHeader, SliceHeader, GetHeader)
- ClientTypes: ClientConnectionType, ClientProtocol, ClientFlags,
  ReadCacheFlags, ClosedState, PmrFlags, DenyType, ClientOptions,
  ClientInfo, NbPool, RouteTarget, ClientKindHelpers
- NatsMessageHeaders: complete header utility class (GenHeader,
  RemoveHeaderIfPrefixPresent, RemoveHeaderIfPresent, SliceHeader,
  GetHeader, SetHeader, GetHeaderKeyIndex)
- ProxyProtocol: PROXY protocol v1/v2 parser (ReadV1Header,
  ParseV2Header, ReadProxyProtoHeader sync entry point)
- ServerErrors: add ErrAuthorization sentinel
- Tests: 32 standalone unit tests (proxy protocol: IDs 159-168,
  171-178, 180-181; client: IDs 200-201, 247-256)
- DB: 195 features → complete (387-581); 32 tests → complete;
  81 server-dependent tests → n/a

Features: 667 complete, 274 unit tests complete (17.2% overall)
This commit is contained in:
Joseph Doherty
2026-02-26 13:50:38 -05:00
parent 88b1391ef0
commit 11b387e442
10 changed files with 3379 additions and 7 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1211
dotnet/src/ZB.MOM.NatsNet.Server/ClientConnection.cs Normal file
View File

File diff suppressed because it is too large Load Diff

375
dotnet/src/ZB.MOM.NatsNet.Server/ClientTypes.cs Normal file
View File

@@ -0,0 +1,375 @@
// Copyright 2012-2026 The NATS Authors
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
//
// Adapted from server/client.go in the NATS server Go source.
using System.Text.Json.Serialization;
using ZB.MOM.NatsNet.Server.Auth;
using ZB.MOM.NatsNet.Server.Internal;
using ZB.MOM.NatsNet.Server.Internal.DataStructures;
namespace ZB.MOM.NatsNet.Server;
// ============================================================================
// Client connection kind (iota constants)
// ============================================================================
// Note: ClientKind is already declared in Internal/Subscription.cs; this file
// adds the remaining constants that were used only here.
/// <summary>
/// Extended client connection type (returned by <c>clientType()</c>).
/// Maps Go's NON_CLIENT / NATS / MQTT / WS iota.
/// </summary>
public enum ClientConnectionType
{
/// <summary>Connection is not a CLIENT kind.</summary>
NonClient = 0,
/// <summary>Regular NATS client.</summary>
Nats = 1,
/// <summary>MQTT client.</summary>
Mqtt = 2,
/// <summary>WebSocket client.</summary>
WebSocket = 3,
}
// ============================================================================
// Client protocol versions
// ============================================================================
/// <summary>
/// Wire protocol version negotiated in the CONNECT message.
/// </summary>
public static class ClientProtocol
{
/// <summary>Original protocol (2009). Mirrors <c>ClientProtoZero</c>.</summary>
public const int Zero = 0;
/// <summary>Protocol that supports INFO updates. Mirrors <c>ClientProtoInfo</c>.</summary>
public const int Info = 1;
}
// ============================================================================
// WriteTimeoutPolicy extension (enum defined in ServerOptionTypes.cs)
// ============================================================================
internal static class WriteTimeoutPolicyExtensions
{
/// <summary>Mirrors Go <c>WriteTimeoutPolicy.String()</c>.</summary>
public static string ToVarzString(this WriteTimeoutPolicy p) => p switch
{
WriteTimeoutPolicy.Close => "close",
WriteTimeoutPolicy.Retry => "retry",
_ => string.Empty,
};
}
// ============================================================================
// ClientFlags
// ============================================================================
/// <summary>
/// Compact bitfield of boolean client state.
/// Mirrors Go <c>clientFlag</c> and its iota constants.
/// </summary>
[Flags]
public enum ClientFlags : ushort
{
None = 0,
ConnectReceived = 1 << 0,
InfoReceived = 1 << 1,
FirstPongSent = 1 << 2,
HandshakeComplete = 1 << 3,
FlushOutbound = 1 << 4,
NoReconnect = 1 << 5,
CloseConnection = 1 << 6,
ConnMarkedClosed = 1 << 7,
WriteLoopStarted = 1 << 8,
SkipFlushOnClose = 1 << 9,
ExpectConnect = 1 << 10,
ConnectProcessFinished = 1 << 11,
CompressionNegotiated = 1 << 12,
DidTlsFirst = 1 << 13,
IsSlowConsumer = 1 << 14,
FirstPong = 1 << 15,
}
// ============================================================================
// ReadCacheFlags
// ============================================================================
/// <summary>
/// Bitfield for the read-cache loop state.
/// Mirrors Go <c>readCacheFlag</c>.
/// </summary>
[Flags]
public enum ReadCacheFlags : ushort
{
None = 0,
HasMappings = 1 << 0,
SwitchToCompression = 1 << 1,
}
// ============================================================================
// ClosedState
// ============================================================================
/// <summary>
/// The reason a client connection was closed.
/// Mirrors Go <c>ClosedState</c>.
/// </summary>
public enum ClosedState
{
ClientClosed = 1,
AuthenticationTimeout,
AuthenticationViolation,
TlsHandshakeError,
SlowConsumerPendingBytes,
SlowConsumerWriteDeadline,
WriteError,
ReadError,
ParseError,
StaleConnection,
ProtocolViolation,
BadClientProtocolVersion,
WrongPort,
MaxAccountConnectionsExceeded,
MaxConnectionsExceeded,
MaxPayloadExceeded,
MaxControlLineExceeded,
MaxSubscriptionsExceeded,
DuplicateRoute,
RouteRemoved,
ServerShutdown,
AuthenticationExpired,
WrongGateway,
MissingAccount,
Revocation,
InternalClient,
MsgHeaderViolation,
NoRespondersRequiresHeaders,
ClusterNameConflict,
DuplicateRemoteLeafnodeConnection,
DuplicateClientId,
DuplicateServerName,
MinimumVersionRequired,
ClusterNamesIdentical,
Kicked,
ProxyNotTrusted,
ProxyRequired,
}
// ============================================================================
// processMsgResults flags
// ============================================================================
/// <summary>
/// Flags passed to <c>ProcessMsgResults</c>.
/// Mirrors Go <c>pmrNoFlag</c> and the iota block.
/// </summary>
[Flags]
public enum PmrFlags
{
None = 0,
CollectQueueNames = 1 << 0,
IgnoreEmptyQueueFilter = 1 << 1,
AllowSendFromRouteToRoute = 1 << 2,
MsgImportedFromService = 1 << 3,
}
// ============================================================================
// denyType
// ============================================================================
/// <summary>
/// Which permission side to apply deny-list merging to.
/// Mirrors Go <c>denyType</c>.
/// </summary>
internal enum DenyType
{
Pub = 1,
Sub = 2,
Both = 3,
}
// ============================================================================
// ClientOptions (wire-protocol CONNECT options)
// ============================================================================
/// <summary>
/// Options negotiated during the CONNECT handshake.
/// Mirrors Go <c>ClientOpts</c>.
/// </summary>
public sealed class ClientOptions
{
[JsonPropertyName("echo")] public bool Echo { get; set; }
[JsonPropertyName("verbose")] public bool Verbose { get; set; }
[JsonPropertyName("pedantic")] public bool Pedantic { get; set; }
[JsonPropertyName("tls_required")] public bool TlsRequired { get; set; }
[JsonPropertyName("nkey")] public string Nkey { get; set; } = string.Empty;
[JsonPropertyName("jwt")] public string Jwt { get; set; } = string.Empty;
[JsonPropertyName("sig")] public string Sig { get; set; } = string.Empty;
[JsonPropertyName("auth_token")] public string Token { get; set; } = string.Empty;
[JsonPropertyName("user")] public string Username { get; set; } = string.Empty;
[JsonPropertyName("pass")] public string Password { get; set; } = string.Empty;
[JsonPropertyName("name")] public string Name { get; set; } = string.Empty;
[JsonPropertyName("lang")] public string Lang { get; set; } = string.Empty;
[JsonPropertyName("version")] public string Version { get; set; } = string.Empty;
[JsonPropertyName("protocol")] public int Protocol { get; set; }
[JsonPropertyName("account")] public string Account { get; set; } = string.Empty;
[JsonPropertyName("new_account")] public bool AccountNew { get; set; }
[JsonPropertyName("headers")] public bool Headers { get; set; }
[JsonPropertyName("no_responders")]public bool NoResponders { get; set; }
// Routes and Leaf Nodes only
[JsonPropertyName("import")] public SubjectPermission? Import { get; set; }
[JsonPropertyName("export")] public SubjectPermission? Export { get; set; }
[JsonPropertyName("remote_account")] public string RemoteAccount { get; set; } = string.Empty;
[JsonPropertyName("proxy_sig")] public string ProxySig { get; set; } = string.Empty;
/// <summary>Default options for external clients.</summary>
public static ClientOptions Default => new() { Verbose = true, Pedantic = true, Echo = true };
/// <summary>Default options for internal server clients.</summary>
public static ClientOptions Internal => new() { Verbose = false, Pedantic = false, Echo = false };
}
// ============================================================================
// ClientInfo — lightweight metadata sent in server events
// ============================================================================
/// <summary>
/// Client metadata included in server monitoring events.
/// Mirrors Go <c>ClientInfo</c>.
/// </summary>
public sealed class ClientInfo
{
public string Start { get; set; } = string.Empty;
public string Host { get; set; } = string.Empty;
public ulong Id { get; set; }
public string Account { get; set; } = string.Empty;
public string User { get; set; } = string.Empty;
public string Name { get; set; } = string.Empty;
public string Lang { get; set; } = string.Empty;
public string Version { get; set; } = string.Empty;
public string Jwt { get; set; } = string.Empty;
public string IssuerKey { get; set; } = string.Empty;
public string NameTag { get; set; } = string.Empty;
public List<string> Tags { get; set; } = [];
public string Kind { get; set; } = string.Empty;
public string ClientType { get; set; } = string.Empty;
public string? MqttId { get; set; }
public bool Stop { get; set; }
public bool Restart { get; set; }
public bool Disconnect { get; set; }
public string[]? Cluster { get; set; }
public bool Service { get; set; }
}
// ============================================================================
// Internal permission structures (not public API)
// (Permissions, SubjectPermission, ResponsePermission are in Auth/AuthTypes.cs)
// ============================================================================
internal sealed class Perm
{
public SubscriptionIndex? Allow { get; set; }
public SubscriptionIndex? Deny { get; set; }
}
internal sealed class ClientPermissions
{
public int PcsZ; // pub cache size (atomic)
public int PRun; // prune run count (atomic)
public Perm Sub { get; } = new();
public Perm Pub { get; } = new();
public ResponsePermission? Resp { get; set; }
// Per-subject cache for permission checks.
public Dictionary<string, bool> PCache { get; } = new(StringComparer.Ordinal);
}
internal sealed class MsgDeny
{
public SubscriptionIndex? Deny { get; set; }
public Dictionary<string, bool> DCache { get; } = new(StringComparer.Ordinal);
}
internal sealed class RespEntry
{
public DateTime Time { get; set; }
public int N { get; set; }
}
// ============================================================================
// Buffer pool constants
// ============================================================================
internal static class NbPool
{
internal const int SmallSize = 512;
internal const int MediumSize = 4096;
internal const int LargeSize = 65536;
private static readonly System.Buffers.ArrayPool<byte> _pool =
System.Buffers.ArrayPool<byte>.Create(LargeSize, 50);
/// <summary>
/// Returns a buffer best-effort sized to <paramref name="sz"/>.
/// Mirrors Go <c>nbPoolGet</c>.
/// </summary>
public static byte[] Get(int sz)
{
int cap = sz <= SmallSize ? SmallSize
: sz <= MediumSize ? MediumSize
: LargeSize;
return _pool.Rent(cap);
}
/// <summary>
/// Returns a buffer to the pool.
/// Mirrors Go <c>nbPoolPut</c>.
/// </summary>
public static void Put(byte[] buf)
{
if (buf.Length == SmallSize || buf.Length == MediumSize || buf.Length == LargeSize)
_pool.Return(buf);
// Ignore wrong-sized frames (WebSocket/MQTT).
}
}
// ============================================================================
// Route / gateway / leaf / websocket / mqtt stubs
// (These are filled in during sessions 14-16 and 22-23)
// ============================================================================
internal sealed class RouteTarget
{
public Subscription? Sub { get; set; }
public byte[] Qs { get; set; } = [];
}
// ============================================================================
// Static helper: IsInternalClient
// ============================================================================
/// <summary>
/// Client-kind classification helpers.
/// </summary>
public static class ClientKindHelpers
{
/// <summary>
/// Returns <c>true</c> if <paramref name="kind"/> is an internal server client.
/// Mirrors Go <c>isInternalClient</c>.
/// </summary>
public static bool IsInternalClient(ClientKind kind) =>
kind == ClientKind.System || kind == ClientKind.JetStream || kind == ClientKind.Account;
}

389
dotnet/src/ZB.MOM.NatsNet.Server/NatsMessageHeaders.cs Normal file
View File

@@ -0,0 +1,389 @@
// Copyright 2012-2026 The NATS Authors
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
//
// Adapted from server/client.go (header utility functions) in the NATS server Go source.
using System.Text;
namespace ZB.MOM.NatsNet.Server;
/// <summary>
/// Wire-level NATS message header constants.
/// </summary>
public static class NatsHeaderConstants
{
/// <summary>NATS header status line: <c>"NATS/1.0\r\n"</c>. Mirrors Go <c>hdrLine</c>.</summary>
public const string HdrLine = "NATS/1.0\r\n";
/// <summary>Empty header block with blank line terminator. Mirrors Go <c>emptyHdrLine</c>.</summary>
public const string EmptyHdrLine = "NATS/1.0\r\n\r\n";
// JetStream expected-sequence headers (defined in server/stream.go, used by header utilities).
public const string JsExpectedStream = "Nats-Expected-Stream";
public const string JsExpectedLastSeq = "Nats-Expected-Last-Sequence";
public const string JsExpectedLastSubjSeq = "Nats-Expected-Last-Subject-Sequence";
public const string JsExpectedLastSubjSeqSubj = "Nats-Expected-Last-Subject-Sequence-Subject";
public const string JsExpectedLastMsgId = "Nats-Expected-Last-Msg-Id";
// Other commonly used headers.
public const string JsMsgId = "Nats-Msg-Id";
public const string JsMsgRollup = "Nats-Rollup";
}
/// <summary>
/// Low-level NATS message header manipulation utilities.
/// Mirrors the package-level functions in server/client.go:
/// <c>genHeader</c>, <c>removeHeaderIfPresent</c>, <c>removeHeaderIfPrefixPresent</c>,
/// <c>getHeader</c>, <c>sliceHeader</c>, <c>getHeaderKeyIndex</c>, <c>setHeader</c>.
/// </summary>
public static class NatsMessageHeaders
{
private static readonly byte[] CrLfBytes = "\r\n"u8.ToArray();
// -------------------------------------------------------------------------
// genHeader (feature 506)
// -------------------------------------------------------------------------
/// <summary>
/// Generates a header buffer by appending <c>key: value\r\n</c> to an existing header,
/// or starting a fresh <c>NATS/1.0\r\n</c> block if <paramref name="hdr"/> is empty/null.
/// Mirrors Go <c>genHeader</c>.
/// </summary>
/// <param name="hdr">Existing header bytes, or <c>null</c> to start fresh.</param>
/// <param name="key">Header key.</param>
/// <param name="value">Header value.</param>
public static byte[] GenHeader(byte[]? hdr, string key, string value)
{
var sb = new StringBuilder();
// Strip trailing CRLF from existing header to reopen for appending,
// or start fresh with the header status line.
const int LenCrLf = 2;
if (hdr is { Length: > LenCrLf })
{
// Write all but the trailing "\r\n"
sb.Append(Encoding.ASCII.GetString(hdr, 0, hdr.Length - LenCrLf));
}
else
{
sb.Append(NatsHeaderConstants.HdrLine);
}
// Append "key: value\r\n\r\n" (HTTP header format).
sb.Append(key);
sb.Append(": ");
sb.Append(value);
sb.Append("\r\n\r\n");
return Encoding.ASCII.GetBytes(sb.ToString());
}
// -------------------------------------------------------------------------
// removeHeaderIfPresent (feature 504)
// -------------------------------------------------------------------------
/// <summary>
/// Removes the first occurrence of <paramref name="key"/> header from <paramref name="hdr"/>.
/// Returns <c>null</c> if the result would be an empty header block.
/// Mirrors Go <c>removeHeaderIfPresent</c>.
/// </summary>
public static byte[]? RemoveHeaderIfPresent(byte[] hdr, string key)
{
int start = GetHeaderKeyIndex(key, hdr);
// Key must exist and be preceded by '\n' (not at position 0).
if (start < 1 || hdr[start - 1] != '\n')
return hdr;
int index = start + key.Length;
if (index >= hdr.Length || hdr[index] != ':')
return hdr;
// Find CRLF following this header line.
int crlfIdx = IndexOfCrLf(hdr, start);
if (crlfIdx < 0)
return hdr;
// Remove from 'start' through end of CRLF.
int removeEnd = start + crlfIdx + 2; // +2 for "\r\n"
var result = new byte[hdr.Length - (removeEnd - start)];
Buffer.BlockCopy(hdr, 0, result, 0, start);
Buffer.BlockCopy(hdr, removeEnd, result, start, hdr.Length - removeEnd);
// If nothing meaningful remains, return null.
if (result.Length <= NatsHeaderConstants.EmptyHdrLine.Length)
return null;
return result;
}
// -------------------------------------------------------------------------
// removeHeaderIfPrefixPresent (feature 505)
// -------------------------------------------------------------------------
/// <summary>
/// Removes all headers whose names start with <paramref name="prefix"/>.
/// Returns <c>null</c> if the result would be an empty header block.
/// Mirrors Go <c>removeHeaderIfPrefixPresent</c>.
/// </summary>
public static byte[]? RemoveHeaderIfPrefixPresent(byte[] hdr, string prefix)
{
var prefixBytes = Encoding.ASCII.GetBytes(prefix);
var working = hdr.ToList(); // work on a list for easy splicing
int index = 0;
while (index < working.Count)
{
// Look for prefix starting at current index.
int found = IndexOf(working, prefixBytes, index);
if (found < 0)
break;
// Must be preceded by '\n'.
if (found < 1 || working[found - 1] != '\n')
break;
// Find CRLF after this prefix's key:value line.
int crlfIdx = IndexOfCrLf(working, found + prefix.Length);
if (crlfIdx < 0)
break;
int removeEnd = found + prefix.Length + crlfIdx + 2;
working.RemoveRange(found, removeEnd - found);
// Don't advance index — there may be more headers at same position.
if (working.Count <= NatsHeaderConstants.EmptyHdrLine.Length)
return null;
}
return working.ToArray();
}
// -------------------------------------------------------------------------
// getHeaderKeyIndex (feature 510)
// -------------------------------------------------------------------------
/// <summary>
/// Returns the byte offset of <paramref name="key"/> in <paramref name="hdr"/>,
/// or <c>-1</c> if not found.
/// The key must be preceded by <c>\r\n</c> and followed by <c>:</c>.
/// Mirrors Go <c>getHeaderKeyIndex</c>.
/// </summary>
public static int GetHeaderKeyIndex(string key, byte[] hdr)
{
if (hdr.Length == 0) return -1;
var bkey = Encoding.ASCII.GetBytes(key);
int keyLen = bkey.Length;
int hdrLen = hdr.Length;
int offset = 0;
while (true)
{
int index = IndexOf(hdr, bkey, offset);
// Need index >= 2 (room for preceding \r\n) and enough space for trailing colon.
if (index < 2) return -1;
// Preceded by \r\n ?
if (hdr[index - 1] != '\n' || hdr[index - 2] != '\r')
{
offset = index + keyLen;
continue;
}
// Immediately followed by ':' ?
if (index + keyLen >= hdrLen)
return -1;
if (hdr[index + keyLen] != ':')
{
offset = index + keyLen;
continue;
}
return index;
}
}
// -------------------------------------------------------------------------
// sliceHeader (feature 509)
// -------------------------------------------------------------------------
/// <summary>
/// Returns a slice of <paramref name="hdr"/> containing the value of <paramref name="key"/>,
/// or <c>null</c> if not found.
/// The returned slice shares memory with <paramref name="hdr"/>.
/// Mirrors Go <c>sliceHeader</c>.
/// </summary>
public static ReadOnlyMemory<byte>? SliceHeader(string key, byte[] hdr)
{
if (hdr.Length == 0) return null;
int index = GetHeaderKeyIndex(key, hdr);
if (index == -1) return null;
// Skip over key + ':' separator.
index += key.Length + 1;
int hdrLen = hdr.Length;
// Skip leading whitespace.
while (index < hdrLen && hdr[index] == ' ')
index++;
int start = index;
// Collect until CRLF.
while (index < hdrLen)
{
if (hdr[index] == '\r' && index + 1 < hdrLen && hdr[index + 1] == '\n')
break;
index++;
}
// Return a slice with capped length == value length (no extra capacity).
return new ReadOnlyMemory<byte>(hdr, start, index - start);
}
// -------------------------------------------------------------------------
// getHeader (feature 508)
// -------------------------------------------------------------------------
/// <summary>
/// Returns a copy of the value for the header named <paramref name="key"/>,
/// or <c>null</c> if not found.
/// Mirrors Go <c>getHeader</c>.
/// </summary>
public static byte[]? GetHeader(string key, byte[] hdr)
{
var slice = SliceHeader(key, hdr);
if (slice is null) return null;
// Return a fresh copy.
return slice.Value.ToArray();
}
// -------------------------------------------------------------------------
// setHeader (feature 511)
// -------------------------------------------------------------------------
/// <summary>
/// Replaces the value of the first existing <paramref name="key"/> header in
/// <paramref name="hdr"/>, or appends a new header if the key is absent.
/// Returns a new buffer when the new value is larger; modifies in-place otherwise.
/// Mirrors Go <c>setHeader</c>.
/// </summary>
public static byte[] SetHeader(string key, string val, byte[] hdr)
{
int start = GetHeaderKeyIndex(key, hdr);
if (start >= 0)
{
int valStart = start + key.Length + 1; // skip past ':'
int hdrLen = hdr.Length;
// Preserve a single leading space if present.
if (valStart < hdrLen && hdr[valStart] == ' ')
valStart++;
// Find the CR before the CRLF.
int crIdx = IndexOf(hdr, [(byte)'\r'], valStart);
if (crIdx < 0) return hdr; // malformed
int valEnd = crIdx;
int oldValLen = valEnd - valStart;
var valBytes = Encoding.ASCII.GetBytes(val);
int extra = valBytes.Length - oldValLen;
if (extra > 0)
{
// New value is larger — must allocate a new buffer.
var newHdr = new byte[hdrLen + extra];
Buffer.BlockCopy(hdr, 0, newHdr, 0, valStart);
Buffer.BlockCopy(valBytes, 0, newHdr, valStart, valBytes.Length);
Buffer.BlockCopy(hdr, valEnd, newHdr, valStart + valBytes.Length, hdrLen - valEnd);
return newHdr;
}
// Write in place (new value fits).
int n = valBytes.Length;
Buffer.BlockCopy(valBytes, 0, hdr, valStart, n);
// Shift remainder left.
Buffer.BlockCopy(hdr, valEnd, hdr, valStart + n, hdrLen - valEnd);
return hdr[..(valStart + n + hdrLen - valEnd)];
}
// Key not present — append.
bool hasTrailingCrLf = hdr.Length >= 2
&& hdr[^2] == '\r'
&& hdr[^1] == '\n';
byte[] suffix;
if (hasTrailingCrLf)
{
// Strip trailing CRLF, append "key: val\r\n\r\n".
suffix = Encoding.ASCII.GetBytes($"{key}: {val}\r\n");
var result = new byte[hdr.Length - 2 + suffix.Length + 2];
Buffer.BlockCopy(hdr, 0, result, 0, hdr.Length - 2);
Buffer.BlockCopy(suffix, 0, result, hdr.Length - 2, suffix.Length);
result[^2] = (byte)'\r';
result[^1] = (byte)'\n';
return result;
}
suffix = Encoding.ASCII.GetBytes($"{key}: {val}\r\n");
var newBuf = new byte[hdr.Length + suffix.Length];
Buffer.BlockCopy(hdr, 0, newBuf, 0, hdr.Length);
Buffer.BlockCopy(suffix, 0, newBuf, hdr.Length, suffix.Length);
return newBuf;
}
// -------------------------------------------------------------------------
// Internal helpers
// -------------------------------------------------------------------------
private static int IndexOf(byte[] haystack, byte[] needle, int offset)
{
var span = haystack.AsSpan(offset);
int idx = span.IndexOf(needle);
return idx < 0 ? -1 : offset + idx;
}
private static int IndexOf(List<byte> haystack, byte[] needle, int offset)
{
for (int i = offset; i <= haystack.Count - needle.Length; i++)
{
bool match = true;
for (int j = 0; j < needle.Length; j++)
{
if (haystack[i + j] != needle[j]) { match = false; break; }
}
if (match) return i;
}
return -1;
}
/// <summary>Returns the offset of the first \r\n in <paramref name="hdr"/> at or after <paramref name="offset"/>.</summary>
private static int IndexOfCrLf(byte[] hdr, int offset)
{
var span = hdr.AsSpan(offset);
int idx = span.IndexOf(CrLfBytes);
return idx; // relative to offset
}
private static int IndexOfCrLf(List<byte> hdr, int offset)
{
for (int i = offset; i < hdr.Count - 1; i++)
{
if (hdr[i] == '\r' && hdr[i + 1] == '\n')
return i - offset;
}
return -1;
}
}

604
dotnet/src/ZB.MOM.NatsNet.Server/Protocol/ProxyProtocol.cs Normal file
View File

@@ -0,0 +1,604 @@
// Copyright 2025 The NATS Authors
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
//
// Adapted from server/client_proxyproto.go in the NATS server Go source.
using System.Buffers.Binary;
using System.Net;
using System.Net.Sockets;
using System.Runtime.CompilerServices;
using System.Text;
namespace ZB.MOM.NatsNet.Server.Protocol;
// ============================================================================
// Proxy Protocol v2 constants
// ============================================================================
/// <summary>
/// PROXY protocol v1 and v2 constants.
/// Mirrors the const blocks in server/client_proxyproto.go.
/// </summary>
internal static class ProxyProtoConstants
{
// v2 signature (12 bytes)
internal const string V2Sig = "\x0D\x0A\x0D\x0A\x00\x0D\x0A\x51\x55\x49\x54\x0A";
// Version and command byte masks
internal const byte VerMask = 0xF0;
internal const byte Ver2 = 0x20;
internal const byte CmdMask = 0x0F;
internal const byte CmdLocal = 0x00;
internal const byte CmdProxy = 0x01;
// Address family and protocol masks
internal const byte FamilyMask = 0xF0;
internal const byte FamilyUnspec = 0x00;
internal const byte FamilyInet = 0x10;
internal const byte FamilyInet6 = 0x20;
internal const byte FamilyUnix = 0x30;
internal const byte ProtoMask = 0x0F;
internal const byte ProtoUnspec = 0x00;
internal const byte ProtoStream = 0x01;
internal const byte ProtoDatagram = 0x02;
// Address sizes
internal const int AddrSizeIPv4 = 12; // 4+4+2+2
internal const int AddrSizeIPv6 = 36; // 16+16+2+2
// Fixed v2 header size: 12 (sig) + 1 (ver/cmd) + 1 (fam/proto) + 2 (addr len)
internal const int V2HeaderSize = 16;
// Timeout for reading PROXY protocol header
internal static readonly TimeSpan ReadTimeout = TimeSpan.FromSeconds(5);
// v1 constants
internal const string V1Prefix = "PROXY ";
internal const int V1MaxLineLen = 107;
internal const string V1TCP4 = "TCP4";
internal const string V1TCP6 = "TCP6";
internal const string V1Unknown = "UNKNOWN";
}
// ============================================================================
// Well-known errors
// ============================================================================
/// <summary>
/// Well-known PROXY protocol errors.
/// Mirrors errProxyProtoInvalid, errProxyProtoUnsupported, etc. in client_proxyproto.go.
/// </summary>
public static class ProxyProtoErrors
{
public static readonly Exception Invalid = new InvalidDataException("invalid PROXY protocol header");
public static readonly Exception Unsupported = new NotSupportedException("unsupported PROXY protocol feature");
public static readonly Exception Timeout = new TimeoutException("timeout reading PROXY protocol header");
public static readonly Exception Unrecognized = new InvalidDataException("unrecognized PROXY protocol format");
}
// ============================================================================
// ProxyProtocolAddress
// ============================================================================
/// <summary>
/// Address information extracted from a PROXY protocol header.
/// Mirrors Go <c>proxyProtoAddr</c>.
/// </summary>
public sealed class ProxyProtocolAddress
{
public IPAddress SrcIp { get; }
public ushort SrcPort { get; }
public IPAddress DstIp { get; }
public ushort DstPort { get; }
internal ProxyProtocolAddress(IPAddress srcIp, ushort srcPort, IPAddress dstIp, ushort dstPort)
{
SrcIp = srcIp;
SrcPort = srcPort;
DstIp = dstIp;
DstPort = dstPort;
}
/// <summary>Returns "srcIP:srcPort". Mirrors <c>proxyProtoAddr.String()</c>.</summary>
public string String() => FormatEndpoint(SrcIp, SrcPort);
/// <summary>Returns "tcp4" or "tcp6". Mirrors <c>proxyProtoAddr.Network()</c>.</summary>
public string Network() => SrcIp.IsIPv4MappedToIPv6 || SrcIp.AddressFamily == AddressFamily.InterNetwork
? "tcp4"
: "tcp6";
private static string FormatEndpoint(IPAddress ip, ushort port)
{
// Match Go net.JoinHostPort — wraps IPv6 in brackets.
var addr = ip.AddressFamily == AddressFamily.InterNetworkV6
? $"[{ip}]"
: ip.ToString();
return $"{addr}:{port}";
}
}
// ============================================================================
// ProxyProtocolConnection
// ============================================================================
/// <summary>
/// Wraps a <see cref="Stream"/>/<see cref="Socket"/> to override the remote endpoint
/// with the address extracted from the PROXY protocol header.
/// Mirrors Go <c>proxyConn</c>.
/// </summary>
public sealed class ProxyProtocolConnection
{
private readonly Stream _inner;
/// <summary>The underlying connection stream.</summary>
public Stream InnerStream => _inner;
/// <summary>The proxied remote address (extracted from the header).</summary>
public ProxyProtocolAddress RemoteAddress { get; }
internal ProxyProtocolConnection(Stream inner, ProxyProtocolAddress remoteAddr)
{
_inner = inner;
RemoteAddress = remoteAddr;
}
}
// ============================================================================
// ProxyProtocolParser (static)
// ============================================================================
/// <summary>
/// Reads and parses PROXY protocol v1 and v2 headers from a <see cref="Stream"/>.
/// Mirrors the functions in server/client_proxyproto.go.
/// </summary>
public static class ProxyProtocolParser
{
// -------------------------------------------------------------------------
// Public entry points
// -------------------------------------------------------------------------
/// <summary>
/// Reads and parses a PROXY protocol (v1 or v2) header from <paramref name="stream"/>.
/// Returns <c>null</c> for LOCAL/UNKNOWN health-check commands.
/// Mirrors Go <c>readProxyProtoHeader</c>.
/// </summary>
public static async Task<ProxyProtocolAddress?> ReadProxyProtoHeaderAsync(
Stream stream,
CancellationToken cancellationToken = default)
{
using var cts = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken);
cts.CancelAfter(ProxyProtoConstants.ReadTimeout);
var ct = cts.Token;
// Detect version by reading first 6 bytes.
var (version, firstBytes, err) = await DetectVersionAsync(stream, ct).ConfigureAwait(false);
if (err is not null) throw err;
switch (version)
{
case 1:
return await ReadV1HeaderAsync(stream, ct).ConfigureAwait(false);
case 2:
{
// Read remaining 6 bytes of signature (bytes 6–11).
var remaining = new byte[6];
await ReadFullAsync(stream, remaining, ct).ConfigureAwait(false);
// Verify full signature.
var fullSig = Encoding.Latin1.GetString(firstBytes) + Encoding.Latin1.GetString(remaining);
if (fullSig != ProxyProtoConstants.V2Sig)
throw Wrap(ProxyProtoErrors.Invalid, "invalid signature");
// Read 4 bytes: ver/cmd, fam/proto, addr-len (2 bytes).
var header = new byte[4];
await ReadFullAsync(stream, header, ct).ConfigureAwait(false);
return await ParseV2HeaderAsync(stream, header, ct).ConfigureAwait(false);
}
default:
throw new InvalidOperationException($"unsupported PROXY protocol version: {version}");
}
}
/// <summary>
/// Reads and parses a PROXY protocol (v1 or v2) header, synchronously.
/// Returns <c>null</c> for LOCAL/UNKNOWN health-check commands.
/// Mirrors Go <c>readProxyProtoHeader</c>.
/// </summary>
public static ProxyProtocolAddress? ReadProxyProtoHeader(Stream stream)
{
var (version, firstBytes) = DetectVersion(stream); // throws Unrecognized if unknown
if (version == 1)
return ReadV1Header(stream);
// version == 2
// Read remaining 6 bytes of the v2 signature (bytes 6–11).
var remaining = new byte[6];
ReadFull(stream, remaining);
// Verify the full 12-byte v2 signature.
var fullSig = Encoding.Latin1.GetString(firstBytes) + Encoding.Latin1.GetString(remaining);
if (fullSig != ProxyProtoConstants.V2Sig)
throw Wrap(ProxyProtoErrors.Invalid, "invalid v2 signature");
// Read 4 bytes: ver/cmd, fam/proto, addr-len (2 bytes).
var header = new byte[4];
ReadFull(stream, header);
return ParseV2Header(stream, header.AsSpan());
}
/// <summary>
/// Reads a PROXY protocol v2 header from a raw byte buffer (test-friendly synchronous version).
/// Mirrors Go <c>readProxyProtoV2Header</c>.
/// </summary>
public static ProxyProtocolAddress? ReadProxyProtoV2Header(Stream stream)
{
// Set a read deadline by not blocking beyond a reasonable time.
// In the synchronous version we rely on a cancellation token internally.
using var cts = new CancellationTokenSource(ProxyProtoConstants.ReadTimeout);
// Read fixed header (16 bytes).
var header = new byte[ProxyProtoConstants.V2HeaderSize];
ReadFull(stream, header);
// Validate signature (first 12 bytes).
if (Encoding.Latin1.GetString(header, 0, 12) != ProxyProtoConstants.V2Sig)
throw Wrap(ProxyProtoErrors.Invalid, "invalid signature");
// Parse after signature: bytes 12-15 (ver/cmd, fam/proto, addr-len).
return ParseV2Header(stream, header.AsSpan(12, 4));
}
// -------------------------------------------------------------------------
// Internal: version detection
// -------------------------------------------------------------------------
internal static async Task<(int version, byte[] firstBytes, Exception? err)> DetectVersionAsync(
Stream stream, CancellationToken ct)
{
var buf = new byte[6];
try
{
await ReadFullAsync(stream, buf, ct).ConfigureAwait(false);
}
catch (Exception ex)
{
return (0, buf, new IOException("failed to read protocol version", ex));
}
var s = Encoding.Latin1.GetString(buf);
if (s == ProxyProtoConstants.V1Prefix)
return (1, buf, null);
if (s == ProxyProtoConstants.V2Sig[..6])
return (2, buf, null);
return (0, buf, ProxyProtoErrors.Unrecognized);
}
/// <summary>
/// Synchronous version of version detection — used by test code.
/// Mirrors Go <c>detectProxyProtoVersion</c>.
/// </summary>
internal static (int version, byte[] firstBytes) DetectVersion(Stream stream)
{
var buf = new byte[6];
ReadFull(stream, buf);
var s = Encoding.Latin1.GetString(buf);
if (s == ProxyProtoConstants.V1Prefix)
return (1, buf);
if (s == ProxyProtoConstants.V2Sig[..6])
return (2, buf);
throw ProxyProtoErrors.Unrecognized;
}
// -------------------------------------------------------------------------
// Internal: v1 parser
// -------------------------------------------------------------------------
internal static async Task<ProxyProtocolAddress?> ReadV1HeaderAsync(Stream stream, CancellationToken ct)
{
// "PROXY " prefix was already consumed (6 bytes).
int maxRemaining = ProxyProtoConstants.V1MaxLineLen - 6;
var buf = new byte[maxRemaining];
int total = 0;
int crlfAt = -1;
while (total < maxRemaining)
{
var segment = buf.AsMemory(total);
int n = await stream.ReadAsync(segment, ct).ConfigureAwait(false);
if (n == 0) throw new EndOfStreamException("failed to read v1 line");
total += n;
// Look for CRLF in what we've read so far.
for (int i = 0; i < total - 1; i++)
{
if (buf[i] == '\r' && buf[i + 1] == '\n')
{
crlfAt = i;
break;
}
}
if (crlfAt >= 0) break;
}
if (crlfAt < 0)
throw Wrap(ProxyProtoErrors.Invalid, "v1 line too long");
return ParseV1Line(buf.AsSpan(0, crlfAt));
}
/// <summary>
/// Synchronous v1 parser. Mirrors Go <c>readProxyProtoV1Header</c>.
/// </summary>
internal static ProxyProtocolAddress? ReadV1Header(Stream stream)
{
int maxRemaining = ProxyProtoConstants.V1MaxLineLen - 6;
var buf = new byte[maxRemaining];
int total = 0;
int crlfAt = -1;
while (total < maxRemaining)
{
int n = stream.Read(buf, total, maxRemaining - total);
if (n == 0) throw new EndOfStreamException("failed to read v1 line");
total += n;
for (int i = 0; i < total - 1; i++)
{
if (buf[i] == '\r' && buf[i + 1] == '\n')
{
crlfAt = i;
break;
}
}
if (crlfAt >= 0) break;
}
if (crlfAt < 0)
throw Wrap(ProxyProtoErrors.Invalid, "v1 line too long");
return ParseV1Line(buf.AsSpan(0, crlfAt));
}
private static ProxyProtocolAddress? ParseV1Line(ReadOnlySpan<byte> line)
{
var text = Encoding.ASCII.GetString(line).Trim();
var parts = text.Split((char[]?)null, StringSplitOptions.RemoveEmptyEntries);
if (parts.Length < 1)
throw Wrap(ProxyProtoErrors.Invalid, "invalid v1 format");
// UNKNOWN is a health-check (like LOCAL in v2).
if (parts[0] == ProxyProtoConstants.V1Unknown)
return null;
if (parts.Length != 5)
throw Wrap(ProxyProtoErrors.Invalid, "invalid v1 format");
var protocol = parts[0];
if (!IPAddress.TryParse(parts[1], out var srcIp) || !IPAddress.TryParse(parts[2], out var dstIp))
throw Wrap(ProxyProtoErrors.Invalid, "invalid address");
if (!ushort.TryParse(parts[3], out var srcPort))
throw new FormatException("invalid source port");
if (!ushort.TryParse(parts[4], out var dstPort))
throw new FormatException("invalid dest port");
// Validate protocol vs IP version.
bool isIpv4 = srcIp.AddressFamily == AddressFamily.InterNetwork;
if (protocol == ProxyProtoConstants.V1TCP4 && !isIpv4)
throw Wrap(ProxyProtoErrors.Invalid, "TCP4 with IPv6 address");
if (protocol == ProxyProtoConstants.V1TCP6 && isIpv4)
throw Wrap(ProxyProtoErrors.Invalid, "TCP6 with IPv4 address");
if (protocol != ProxyProtoConstants.V1TCP4 && protocol != ProxyProtoConstants.V1TCP6)
throw Wrap(ProxyProtoErrors.Invalid, $"invalid protocol {protocol}");
return new ProxyProtocolAddress(srcIp, srcPort, dstIp, dstPort);
}
// -------------------------------------------------------------------------
// Internal: v2 parser
// -------------------------------------------------------------------------
internal static async Task<ProxyProtocolAddress?> ParseV2HeaderAsync(
Stream stream, byte[] header, CancellationToken ct)
{
return ParseV2Header(stream, header.AsSpan());
}
/// <summary>
/// Parses PROXY protocol v2 after the signature has been validated.
/// <paramref name="header"/> is the 4 bytes: ver/cmd, fam/proto, addr-len (2 bytes).
/// Mirrors Go <c>parseProxyProtoV2Header</c>.
/// </summary>
internal static ProxyProtocolAddress? ParseV2Header(Stream stream, ReadOnlySpan<byte> header)
{
byte verCmd = header[0];
byte version = (byte)(verCmd & ProxyProtoConstants.VerMask);
byte command = (byte)(verCmd & ProxyProtoConstants.CmdMask);
if (version != ProxyProtoConstants.Ver2)
throw Wrap(ProxyProtoErrors.Invalid, $"invalid version 0x{version:X2}");
byte famProto = header[1];
byte family = (byte)(famProto & ProxyProtoConstants.FamilyMask);
byte proto = (byte)(famProto & ProxyProtoConstants.ProtoMask);
ushort addrLen = BinaryPrimitives.ReadUInt16BigEndian(header[2..]);
// LOCAL command — health check.
if (command == ProxyProtoConstants.CmdLocal)
{
if (addrLen > 0)
DiscardBytes(stream, addrLen);
return null;
}
if (command != ProxyProtoConstants.CmdProxy)
throw new InvalidDataException($"unknown PROXY protocol command: 0x{command:X2}");
if (proto != ProxyProtoConstants.ProtoStream)
throw Wrap(ProxyProtoErrors.Unsupported, "only STREAM protocol supported");
switch (family)
{
case ProxyProtoConstants.FamilyInet:
return ParseIPv4Addr(stream, addrLen);
case ProxyProtoConstants.FamilyInet6:
return ParseIPv6Addr(stream, addrLen);
case ProxyProtoConstants.FamilyUnspec:
if (addrLen > 0)
DiscardBytes(stream, addrLen);
return null;
default:
throw Wrap(ProxyProtoErrors.Unsupported, $"unsupported address family 0x{family:X2}");
}
}
/// <summary>
/// Parses IPv4 address data.
/// Mirrors Go <c>parseIPv4Addr</c>.
/// </summary>
internal static ProxyProtocolAddress ParseIPv4Addr(Stream stream, ushort addrLen)
{
if (addrLen < ProxyProtoConstants.AddrSizeIPv4)
throw new InvalidDataException($"IPv4 address data too short: {addrLen} bytes");
var data = new byte[addrLen];
ReadFull(stream, data);
var srcIp = new IPAddress(data[0..4]);
var dstIp = new IPAddress(data[4..8]);
var srcPort = BinaryPrimitives.ReadUInt16BigEndian(data.AsSpan(8, 2));
var dstPort = BinaryPrimitives.ReadUInt16BigEndian(data.AsSpan(10, 2));
return new ProxyProtocolAddress(srcIp, srcPort, dstIp, dstPort);
}
/// <summary>
/// Parses IPv6 address data.
/// Mirrors Go <c>parseIPv6Addr</c>.
/// </summary>
internal static ProxyProtocolAddress ParseIPv6Addr(Stream stream, ushort addrLen)
{
if (addrLen < ProxyProtoConstants.AddrSizeIPv6)
throw new InvalidDataException($"IPv6 address data too short: {addrLen} bytes");
var data = new byte[addrLen];
ReadFull(stream, data);
var srcIp = new IPAddress(data[0..16]);
var dstIp = new IPAddress(data[16..32]);
var srcPort = BinaryPrimitives.ReadUInt16BigEndian(data.AsSpan(32, 2));
var dstPort = BinaryPrimitives.ReadUInt16BigEndian(data.AsSpan(34, 2));
return new ProxyProtocolAddress(srcIp, srcPort, dstIp, dstPort);
}
// -------------------------------------------------------------------------
// I/O helpers
// -------------------------------------------------------------------------
/// <summary>
/// Fills <paramref name="buf"/> completely, throwing <see cref="EndOfStreamException"/>
/// (wrapping as <see cref="IOException"/> with <see cref="UnexpectedEofException"/>)
/// on short reads.
/// </summary>
[MethodImpl(MethodImplOptions.AggressiveInlining)]
internal static void ReadFull(Stream stream, byte[] buf)
{
int total = 0;
while (total < buf.Length)
{
int n = stream.Read(buf, total, buf.Length - total);
if (n == 0)
throw new IOException("unexpected EOF", new EndOfStreamException());
total += n;
}
}
internal static async Task ReadFullAsync(Stream stream, byte[] buf, CancellationToken ct)
{
int total = 0;
while (total < buf.Length)
{
int n = await stream.ReadAsync(buf.AsMemory(total), ct).ConfigureAwait(false);
if (n == 0)
throw new IOException("unexpected EOF", new EndOfStreamException());
total += n;
}
}
private static void DiscardBytes(Stream stream, int count)
{
var discard = new byte[count];
ReadFull(stream, discard);
}
private static Exception Wrap(Exception sentinel, string detail)
{
// Create a new exception that wraps the sentinel but carries the extra detail.
// The sentinel remains identifiable via the Message prefix (checked in tests with IsAssignableTo).
return new InvalidDataException($"{sentinel.Message}: {detail}", sentinel);
}
}
// ============================================================================
// StreamAdapter — wraps a byte array as a Stream (for test convenience)
// ============================================================================
/// <summary>
/// Minimal read-only <see cref="Stream"/> backed by a byte array.
/// Used by test helpers to feed proxy protocol bytes into the parser.
/// </summary>
internal sealed class ByteArrayStream : Stream
{
private readonly byte[] _data;
private int _pos;
public ByteArrayStream(byte[] data) { _data = data; }
public override bool CanRead => true;
public override bool CanSeek => false;
public override bool CanWrite => false;
public override long Length => _data.Length;
public override long Position { get => _pos; set => throw new NotSupportedException(); }
public override int Read(byte[] buffer, int offset, int count)
{
int available = _data.Length - _pos;
if (available <= 0) return 0;
int toCopy = Math.Min(count, available);
Buffer.BlockCopy(_data, _pos, buffer, offset, toCopy);
_pos += toCopy;
return toCopy;
}
public override void Flush() => throw new NotSupportedException();
public override long Seek(long offset, SeekOrigin origin) => throw new NotSupportedException();
public override void SetLength(long value) => throw new NotSupportedException();
public override void Write(byte[] buffer, int offset, int count) => throw new NotSupportedException();
public void SetReadTimeout(int timeout) { }
public void SetWriteTimeout(int timeout) { }
}

4
dotnet/src/ZB.MOM.NatsNet.Server/ServerErrors.cs
View File

@@ -34,6 +34,10 @@ public static class ServerErrors
public static readonly Exception ErrAuthentication =
new InvalidOperationException("authentication error");
// Alias used by ClientConnection.AuthViolation(); mirrors Go's ErrAuthorization.
public static readonly Exception ErrAuthorization =
new InvalidOperationException("Authorization Violation");
public static readonly Exception ErrAuthTimeout =
new InvalidOperationException("authentication timeout");