EonaCat/EonaCat.Connections
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Made it more secure

Browse Source
This commit is contained in:
EonaCat
2025-08-25 19:30:58 +02:00
parent 0e299d0d27
commit 0c659e9d34
6 changed files with 533 additions and 738 deletions
Download Patch File Download Diff File Expand all files Collapse all files

342
EonaCat.Connections/Processors/JsonDataProcessor.cs
View File

@@ -7,63 +7,74 @@ using Timer = System.Timers.Timer;
namespace EonaCat.Connections.Processors
{
// This file is part of the EonaCat project(s) which is released under the Apache License.
// See the LICENSE file or go to https://EonaCat.com/license for full license details.
/// <summary>
/// Processes incoming data streams into JSON or text messages per client buffer.
/// </summary>
public class JsonDataProcessor<TMessage> : IDisposable
{
public int MaxAllowedBufferSize = 20 * 1024 * 1024;
public int MaxMessagesPerBatch = 200;
private readonly ConcurrentDictionary<string, BufferEntry> _buffers = new();
private const int DefaultMaxBufferSize = 20 * 1024 * 1024; // 20 MB
private const int DefaultMaxMessagesPerBatch = 200;
private static readonly TimeSpan DefaultClientBufferTimeout = TimeSpan.FromMinutes(5);
private readonly ConcurrentDictionary<string, BufferEntry> _buffers = new ConcurrentDictionary<string, BufferEntry>();
private readonly Timer _cleanupTimer;
private readonly TimeSpan _clientBufferTimeout = TimeSpan.FromMinutes(5);
private bool _isDisposed;
/// <summary>
/// This clientName will be used for the buffer (if not set in the DataReceivedEventArgs).
/// Maximum allowed buffer size in bytes (default: 20 MB).
/// </summary>
public int MaxAllowedBufferSize { get; set; } = DefaultMaxBufferSize;
/// <summary>
/// Maximum number of messages processed per batch (default: 200).
/// </summary>
public int MaxMessagesPerBatch { get; set; } = DefaultMaxMessagesPerBatch;
/// <summary>
/// Default client name when one is not provided in <see cref="DataReceivedEventArgs"/>.
/// </summary>
public string ClientName { get; set; } = Guid.NewGuid().ToString();
public Action<TMessage, string, string> ProcessMessage { get; set; }
public Action<string, string> ProcessTextMessage { get; set; }
public event EventHandler<Exception> OnMessageError;
public event EventHandler<Exception> OnError;
private class BufferEntry
{
public readonly StringBuilder Buffer = new();
public readonly StringBuilder Buffer = new StringBuilder();
public DateTime LastUsed = DateTime.UtcNow;
public readonly object SyncRoot = new();
public readonly object SyncRoot = new object();
}
public Action<TMessage, string, string>? ProcessMessage;
public Action<string, string>? ProcessTextMessage;
public event EventHandler<Exception>? OnMessageError;
public event EventHandler<Exception>? OnError;
public JsonDataProcessor()
{
_cleanupTimer = new Timer(_clientBufferTimeout.TotalMilliseconds / 5);
_cleanupTimer.Elapsed += CleanupInactiveClients;
_cleanupTimer = new Timer(DefaultClientBufferTimeout.TotalMilliseconds / 5);
_cleanupTimer.AutoReset = true;
_cleanupTimer.Elapsed += CleanupInactiveClients;
_cleanupTimer.Start();
}
/// <summary>
/// Process incoming raw data.
/// </summary>
public void Process(DataReceivedEventArgs e)
{
if (_isDisposed)
{
throw new ObjectDisposedException(nameof(JsonDataProcessor<TMessage>));
}
EnsureNotDisposed();
if (e.IsBinary)
{
e.StringData = Encoding.UTF8.GetString(e.Data);
}
if (string.IsNullOrEmpty(e.StringData))
if (string.IsNullOrWhiteSpace(e.StringData))
{
OnError?.Invoke(this, new Exception("Received empty data."));
return;
}
string clientName = !string.IsNullOrWhiteSpace(e.Nickname) ? e.Nickname : ClientName;
string clientName = string.IsNullOrWhiteSpace(e.Nickname) ? ClientName : e.Nickname;
string incomingText = e.StringData.Trim();
if (incomingText.Length == 0)
{
@@ -71,12 +82,9 @@ namespace EonaCat.Connections.Processors
}
var bufferEntry = _buffers.GetOrAdd(clientName, _ => new BufferEntry());
List<string>? jsonChunksToProcess = null;
string? textMessageToProcess = null;
lock (bufferEntry.SyncRoot)
{
// Prevent growth before appending
if (bufferEntry.Buffer.Length > MaxAllowedBufferSize)
{
bufferEntry.Buffer.Clear();
@@ -87,12 +95,14 @@ namespace EonaCat.Connections.Processors
int processedCount = 0;
while (processedCount < MaxMessagesPerBatch && ExtractNextJson(bufferEntry.Buffer, out var jsonChunk))
while (processedCount < MaxMessagesPerBatch &&
ExtractNextJson(bufferEntry.Buffer, out var jsonChunk))
{
ProcessDataReceived(jsonChunk, clientName);
processedCount++;
}
// Handle leftover non-JSON text
if (bufferEntry.Buffer.Length > 0 && !ContainsJsonStructure(bufferEntry.Buffer))
{
var leftover = bufferEntry.Buffer.ToString();
@@ -100,50 +110,30 @@ namespace EonaCat.Connections.Processors
ProcessTextMessage?.Invoke(leftover, clientName);
}
}
if (textMessageToProcess != null)
{
ProcessTextMessage?.Invoke(textMessageToProcess, clientName);
}
if (jsonChunksToProcess != null)
{
foreach (var jsonChunk in jsonChunksToProcess)
{
ProcessDataReceived(jsonChunk, clientName);
}
}
}
private void ProcessDataReceived(string? data, string clientName)
private void ProcessDataReceived(string data, string clientName)
{
if (_isDisposed)
{
throw new ObjectDisposedException(nameof(JsonDataProcessor<TMessage>));
}
if (data == null)
{
return;
}
if (string.IsNullOrEmpty(clientName))
{
clientName = ClientName;
}
if (string.IsNullOrWhiteSpace(data) || data.Length == 0)
{
return;
}
EnsureNotDisposed();
if (string.IsNullOrWhiteSpace(data))
{
return;
}
if (string.IsNullOrWhiteSpace(clientName))
{
clientName = ClientName;
}
bool looksLikeJson = data.Length > 1 &&
((data[0] == '{' && data[data.Length - 1] == '}') || (data[0] == '[' && data[data.Length - 1] == ']'));
((data[0] == '{' && data[data.Length - 1] == '}') ||
(data[0] == '[' && data[data.Length - 1] == ']') ||
data[0] == '"' || // string
char.IsDigit(data[0]) || data[0] == '-' || // numbers
data.StartsWith("true") ||
data.StartsWith("false") ||
data.StartsWith("null"));
if (!looksLikeJson)
{
@@ -153,34 +143,19 @@ namespace EonaCat.Connections.Processors
try
{
if (data.Contains("Exception") || data.Contains("Error"))
// Try to detect JSON-encoded exceptions
if (data.IndexOf("Exception", StringComparison.OrdinalIgnoreCase) >= 0 ||
data.IndexOf("Error", StringComparison.OrdinalIgnoreCase) >= 0)
{
try
{
var jsonObject = JObject.Parse(data);
var exceptionToken = jsonObject.SelectToken("Exception");
if (exceptionToken is { Type: not JTokenType.Null })
{
var exception = JsonHelper.ExtractException(data);
if (exception != null)
{
var currentException = new Exception(exception.Message);
OnMessageError?.Invoke(this, currentException);
}
}
}
catch (Exception)
{
// Do nothing
}
TryHandleJsonException(data);
}
var messages = JsonHelper.ToObjects<TMessage>(data);
if (messages != null)
if (messages != null && ProcessMessage != null)
{
foreach (var message in messages)
{
ProcessMessage?.Invoke(message, clientName, data);
ProcessMessage(message, clientName, data);
}
}
}
@@ -190,8 +165,28 @@ namespace EonaCat.Connections.Processors
}
}
private void TryHandleJsonException(string data)
{
try
{
var jsonObject = JObject.Parse(data);
var exceptionToken = jsonObject.SelectToken("Exception");
if (exceptionToken != null && exceptionToken.Type != JTokenType.Null)
{
var exception = JsonHelper.ExtractException(data);
if (exception != null && OnMessageError != null)
{
OnMessageError(this, new Exception(exception.Message));
}
}
}
catch
{
// Ignore malformed exception JSON
}
}
private static bool ExtractNextJson(StringBuilder buffer, out string? json)
private static bool ExtractNextJson(StringBuilder buffer, out string json)
{
json = null;
if (buffer.Length == 0)
@@ -200,13 +195,12 @@ namespace EonaCat.Connections.Processors
}
int depth = 0;
bool inString = false;
bool escape = false;
bool inString = false, escape = false;
int startIndex = -1;
for (int i = 0; i < buffer.Length; i++)
{
char currentCharacter = buffer[i];
char c = buffer[i];
if (inString)
{
@@ -214,71 +208,99 @@ namespace EonaCat.Connections.Processors
{
escape = false;
}
else if (currentCharacter == '\\')
else if (c == '\\')
{
escape = true;
}
else if (currentCharacter == '"')
else if (c == '"')
{
inString = false;
}
}
else
{
if (currentCharacter == '"')
switch (c)
{
inString = true;
if (depth == 0 && startIndex == -1)
{
startIndex = i; // string-only JSON
}
}
else if (currentCharacter == '{' || currentCharacter == '[')
{
if (depth == 0)
{
startIndex = i;
}
case '"':
inString = true;
if (depth == 0 && startIndex == -1)
{
startIndex = i; // string-only JSON
}
depth++;
}
else if (currentCharacter == '}' || currentCharacter == ']')
{
depth--;
if (depth == 0 && startIndex != -1)
{
json = buffer.ToString(startIndex, i - startIndex + 1);
buffer.Remove(0, i + 1);
return true;
}
}
else if (depth == 0 && startIndex == -1 &&
(char.IsDigit(currentCharacter) || currentCharacter == '-' || currentCharacter == 't' || currentCharacter == 'f' || currentCharacter == 'n'))
{
startIndex = i;
break;
// Find token end
int tokenEnd = FindPrimitiveEnd(buffer, i);
json = buffer.ToString(startIndex, tokenEnd - startIndex);
buffer.Remove(0, tokenEnd);
return true;
case '{':
case '[':
if (depth == 0)
{
startIndex = i;
}
depth++;
break;
case '}':
case ']':
depth--;
if (depth == 0 && startIndex != -1)
{
int length = i - startIndex + 1;
json = buffer.ToString(startIndex, length);
buffer.Remove(0, i + 1);
return true;
}
break;
default:
if (depth == 0 && startIndex == -1 &&
(char.IsDigit(c) || c == '-' || c == 't' || c == 'f' || c == 'n'))
{
startIndex = i;
int tokenEnd = FindPrimitiveEnd(buffer, i);
json = buffer.ToString(startIndex, tokenEnd - startIndex);
buffer.Remove(0, tokenEnd);
return true;
}
break;
}
}
}
return false;
}
private static int FindPrimitiveEnd(StringBuilder buffer, int startIndex)
{
for (int i = startIndex; i < buffer.Length; i++)
// Keywords: true/false/null
if (buffer.Length >= startIndex + 4 && buffer.ToString(startIndex, 4) == "true")
{
return startIndex + 4;
}
if (buffer.Length >= startIndex + 5 && buffer.ToString(startIndex, 5) == "false")
{
return startIndex + 5;
}
if (buffer.Length >= startIndex + 4 && buffer.ToString(startIndex, 4) == "null")
{
return startIndex + 4;
}
// Numbers: scan until non-number/decimal/exponent
int i = startIndex;
while (i < buffer.Length)
{
char c = buffer[i];
if (char.IsWhiteSpace(c) || c == ',' || c == ']' || c == '}')
if (!(char.IsDigit(c) || c == '-' || c == '+' || c == '.' || c == 'e' || c == 'E'))
{
return i;
break;
}
i++;
}
return buffer.Length;
return i;
}
private static bool ContainsJsonStructure(StringBuilder buffer)
@@ -286,9 +308,7 @@ namespace EonaCat.Connections.Processors
for (int i = 0; i < buffer.Length; i++)
{
char c = buffer[i];
if (c == '{' || c == '[' || c == '"' ||
c == 't' || c == 'f' || c == 'n' ||
c == '-' || char.IsDigit(c))
if (c == '{' || c == '[' || c == '"' || c == 't' || c == 'f' || c == 'n' || c == '-' || char.IsDigit(c))
{
return true;
}
@@ -296,18 +316,22 @@ namespace EonaCat.Connections.Processors
return false;
}
private void CleanupInactiveClients(object? sender, ElapsedEventArgs e)
private void CleanupInactiveClients(object sender, ElapsedEventArgs e)
{
var now = DateTime.UtcNow;
foreach (var kvp in _buffers)
{
var bufferEntry = kvp.Value;
if (now - bufferEntry.LastUsed > _clientBufferTimeout && _buffers.TryRemove(kvp.Key, out var removed))
if (now - bufferEntry.LastUsed > DefaultClientBufferTimeout)
{
lock (removed.SyncRoot)
BufferEntry removed;
if (_buffers.TryRemove(kvp.Key, out removed))
{
removed.Buffer.Clear();
lock (removed.SyncRoot)
{
removed.Buffer.Clear();
}
}
}
}
@@ -320,7 +344,8 @@ namespace EonaCat.Connections.Processors
return;
}
if (_buffers.TryRemove(clientName, out var removed))
BufferEntry removed;
if (_buffers.TryRemove(clientName, out removed))
{
lock (removed.SyncRoot)
{
@@ -329,6 +354,14 @@ namespace EonaCat.Connections.Processors
}
}
private void EnsureNotDisposed()
{
if (_isDisposed)
{
throw new ObjectDisposedException(nameof(JsonDataProcessor<TMessage>));
}
}
public void Dispose()
{
if (_isDisposed)
@@ -336,25 +369,30 @@ namespace EonaCat.Connections.Processors
return;
}
_isDisposed = true;
_cleanupTimer.Stop();
_cleanupTimer.Elapsed -= CleanupInactiveClients;
_cleanupTimer.Dispose();
foreach (var bufferEntry in _buffers.Values)
try
{
lock (bufferEntry.SyncRoot)
{
bufferEntry.Buffer.Clear();
}
}
_buffers.Clear();
_cleanupTimer.Stop();
_cleanupTimer.Elapsed -= CleanupInactiveClients;
_cleanupTimer.Dispose();
ProcessMessage = null;
ProcessTextMessage = null;
OnMessageError = null;
OnError = null;
foreach (var bufferEntry in _buffers.Values)
{
lock (bufferEntry.SyncRoot)
{
bufferEntry.Buffer.Clear();
}
}
_buffers.Clear();
ProcessMessage = null;
ProcessTextMessage = null;
OnMessageError = null;
OnError = null;
}
finally
{
_isDisposed = true;
}
}
}
}