diff --git a/Core/Resgrid.Model/Providers/IRabbitInboundEventProvider.cs b/Core/Resgrid.Model/Providers/IRabbitInboundEventProvider.cs index f0008b2de..f69b504f5 100644 --- a/Core/Resgrid.Model/Providers/IRabbitInboundEventProvider.cs +++ b/Core/Resgrid.Model/Providers/IRabbitInboundEventProvider.cs @@ -7,6 +7,12 @@ namespace Resgrid.Model.Providers public interface IRabbitInboundEventProvider { Task Start(string clientName, string queueName); + + /// + /// True when the consumer channel exists and is open. Hosts poll this to detect a dead + /// consumer (e.g. after the shared connection was force-reset) and re-call Start. + /// + bool IsConnected(); void RegisterForEvents(Func personnelStatusChanged, Func unitStatusChanged, Func callStatusChanged, diff --git a/Core/Resgrid.Services/EncryptionService.cs b/Core/Resgrid.Services/EncryptionService.cs index 18df5624c..53ef29e88 100644 --- a/Core/Resgrid.Services/EncryptionService.cs +++ b/Core/Resgrid.Services/EncryptionService.cs @@ -7,15 +7,28 @@ namespace Resgrid.Services { /// - /// AES-256-CBC encryption service using PBKDF2-HMAC-SHA256 key derivation. + /// AES-256-GCM authenticated encryption service using PBKDF2-HMAC-SHA256 key derivation. /// The PBKDF2 iteration count defaults to 600,000 per OWASP guidance and is /// configurable via . /// Can be used anywhere in the system that requires encryption at rest. + /// + /// Payload format: new ciphertexts are "enc2:" + Base64(nonce | tag | cipher). Inputs without + /// that prefix are decrypted via the legacy AES-256-CBC/PKCS7 path so data already at rest + /// (department credentials such as Twilio AccountSid/AuthToken JSON) keeps decrypting. The + /// prefix is applied to the Base64 string rather than a leading payload byte because ':' can + /// never appear in Base64 output — a legacy ciphertext (whose first bytes are a random IV) + /// can therefore never be misdetected as GCM, and no try-one-then-the-other fallback is + /// needed, which keeps wrong-key GCM failures deterministic (the tag check always throws). + /// Legacy payloads migrate to GCM whenever a caller re-encrypts; the service itself has no + /// storage access, so opportunistic re-encrypt-on-read is a caller concern. /// public class EncryptionService : IEncryptionService { private const int KeySize = 32; // 256-bit - private const int IvSize = 16; // 128-bit block + private const int IvSize = 16; // 128-bit block (legacy CBC) + private const int NonceSize = 12; // GCM standard nonce + private const int TagSize = 16; // GCM authentication tag + private const string GcmPrefix = "enc2:"; public string Encrypt(string plainText) { @@ -53,26 +66,68 @@ public string DecryptForDepartment(string cipherText, int departmentId, string d private static string EncryptWithKey(string plainText, byte[] key) { - using var aes = Aes.Create(); - aes.KeySize = 256; - aes.BlockSize = 128; - aes.Mode = CipherMode.CBC; - aes.Padding = PaddingMode.PKCS7; - aes.GenerateIV(); - - using var encryptor = aes.CreateEncryptor(key, aes.IV); + var nonce = RandomNumberGenerator.GetBytes(NonceSize); var plainBytes = Encoding.UTF8.GetBytes(plainText); - var cipherBytes = encryptor.TransformFinalBlock(plainBytes, 0, plainBytes.Length); + var cipherBytes = new byte[plainBytes.Length]; + var tag = new byte[TagSize]; - // Prepend IV so we can recover it on decryption - var result = new byte[IvSize + cipherBytes.Length]; - Buffer.BlockCopy(aes.IV, 0, result, 0, IvSize); - Buffer.BlockCopy(cipherBytes, 0, result, IvSize, cipherBytes.Length); + using (var aes = new AesGcm(key, TagSize)) + aes.Encrypt(nonce, plainBytes, cipherBytes, tag); - return Convert.ToBase64String(result); + var result = new byte[NonceSize + TagSize + cipherBytes.Length]; + Buffer.BlockCopy(nonce, 0, result, 0, NonceSize); + Buffer.BlockCopy(tag, 0, result, NonceSize, TagSize); + Buffer.BlockCopy(cipherBytes, 0, result, NonceSize + TagSize, cipherBytes.Length); + + return GcmPrefix + Convert.ToBase64String(result); } private static string DecryptWithKey(string cipherText, byte[] key) + { + if (cipherText.StartsWith(GcmPrefix, StringComparison.Ordinal)) + return DecryptGcm(cipherText.Substring(GcmPrefix.Length), key); + + return DecryptLegacyCbc(cipherText, key); + } + + private static string DecryptGcm(string base64Payload, byte[] key) + { + byte[] fullBytes; + try + { + fullBytes = Convert.FromBase64String(base64Payload); + } + catch (FormatException ex) + { + throw new CryptographicException("Cipher text is not valid Base64.", ex); + } + + if (fullBytes.Length < NonceSize + TagSize) + throw new CryptographicException("Cipher text is too short to contain a valid nonce and tag."); + + var nonce = new byte[NonceSize]; + var tag = new byte[TagSize]; + var cipherBytes = new byte[fullBytes.Length - NonceSize - TagSize]; + Buffer.BlockCopy(fullBytes, 0, nonce, 0, NonceSize); + Buffer.BlockCopy(fullBytes, NonceSize, tag, 0, TagSize); + Buffer.BlockCopy(fullBytes, NonceSize + TagSize, cipherBytes, 0, cipherBytes.Length); + + var plainBytes = new byte[cipherBytes.Length]; + + // A wrong key or any ciphertext/tag tampering fails the tag check and throws + // AuthenticationTagMismatchException (a CryptographicException) — deterministically, + // unlike the legacy CBC padding check. + using (var aes = new AesGcm(key, TagSize)) + aes.Decrypt(nonce, cipherBytes, tag, plainBytes); + + return Encoding.UTF8.GetString(plainBytes); + } + + /// + /// Decrypts the pre-"enc2:" AES-256-CBC/PKCS7 format (IV | cipher, Base64). Retained so + /// data encrypted before the GCM migration keeps decrypting; never used for new payloads. + /// + private static string DecryptLegacyCbc(string cipherText, byte[] key) { byte[] fullBytes; try diff --git a/Core/Resgrid.Services/ModerationService.cs b/Core/Resgrid.Services/ModerationService.cs index 2ee38f4db..3594a3d5e 100644 --- a/Core/Resgrid.Services/ModerationService.cs +++ b/Core/Resgrid.Services/ModerationService.cs @@ -200,11 +200,16 @@ await RecordDepartmentAuditAsync(request, AuditLogTypes.ModerationRequestReopene } catch (Exception ex) { - Logging.LogException(ex); + // A unique-constraint failure here is the expected duplicate-report race: another + // submission from the same reporter won the insert. Only log when no concurrent + // winner exists — that's a real failure being rethrown, not the recovered race. var concurrent = await _moderationReportRepository.GetByRequestAndReporterAsync( request.ModerationRequestId, reportedByUserId, useUnitOfWork: false); if (concurrent == null) + { + Logging.LogException(ex); throw; + } _unitOfWork.DiscardChanges(); return concurrent; diff --git a/Providers/Resgrid.Providers.Bus.Rabbit/RabbitConnection.cs b/Providers/Resgrid.Providers.Bus.Rabbit/RabbitConnection.cs index ef5e17219..8cd523887 100644 --- a/Providers/Resgrid.Providers.Bus.Rabbit/RabbitConnection.cs +++ b/Providers/Resgrid.Providers.Bus.Rabbit/RabbitConnection.cs @@ -34,35 +34,46 @@ public static async Task VerifyAndCreateClients(string clientName) try { - _factory = new ConnectionFactory() { HostName = ServiceBusConfig.RabbitHostname, UserName = ServiceBusConfig.RabbitUsername, Password = ServiceBusConfig.RabbbitPassword }; - _connection = await _factory.CreateConnectionAsync(clientName); - } - catch (Exception ex) - { - Logging.LogException(ex); - - if (!String.IsNullOrWhiteSpace(ServiceBusConfig.RabbitHostname2)) + // Re-check inside the lock: concurrent callers (e.g. several publishers recovering + // through ForceResetAsync at once) all pass the unsynchronized null check above and + // queue on the semaphore. Without this re-check each waiter would create its own + // IConnection in turn — orphaning the previous one undisposed and inflating the + // broker's connection count. Only the first acquirer may create the connection. + if (_connection == null) { try { - _factory = new ConnectionFactory() { HostName = ServiceBusConfig.RabbitHostname2, UserName = ServiceBusConfig.RabbitUsername, Password = ServiceBusConfig.RabbbitPassword }; + _factory = new ConnectionFactory() { HostName = ServiceBusConfig.RabbitHostname, UserName = ServiceBusConfig.RabbitUsername, Password = ServiceBusConfig.RabbbitPassword }; _connection = await _factory.CreateConnectionAsync(clientName); } - catch (Exception ex2) + catch (Exception ex) { - Logging.LogException(ex2); + Logging.LogException(ex); - if (!String.IsNullOrWhiteSpace(ServiceBusConfig.RabbitHostname3)) + if (!String.IsNullOrWhiteSpace(ServiceBusConfig.RabbitHostname2)) { try { - _factory = new ConnectionFactory() { HostName = ServiceBusConfig.RabbitHostname3, UserName = ServiceBusConfig.RabbitUsername, Password = ServiceBusConfig.RabbbitPassword }; + _factory = new ConnectionFactory() { HostName = ServiceBusConfig.RabbitHostname2, UserName = ServiceBusConfig.RabbitUsername, Password = ServiceBusConfig.RabbbitPassword }; _connection = await _factory.CreateConnectionAsync(clientName); } - catch (Exception ex3) + catch (Exception ex2) { - Logging.LogException(ex3); - throw; + Logging.LogException(ex2); + + if (!String.IsNullOrWhiteSpace(ServiceBusConfig.RabbitHostname3)) + { + try + { + _factory = new ConnectionFactory() { HostName = ServiceBusConfig.RabbitHostname3, UserName = ServiceBusConfig.RabbitUsername, Password = ServiceBusConfig.RabbbitPassword }; + _connection = await _factory.CreateConnectionAsync(clientName); + } + catch (Exception ex3) + { + Logging.LogException(ex3); + throw; + } + } } } } @@ -239,6 +250,43 @@ await channel.QueueDeclareAsync( } } + /// + /// Disposes the shared connection and clears cached state so the next CreateConnection call + /// builds a fresh one. Needed when the connection is open but unusable — e.g. its channel + /// numbers are exhausted (ChannelAllocationException) — which the IsOpen guards can never + /// detect. Raises ConnectionReset so cached declaration state is cleared. + /// + public static async Task ForceResetAsync() + { + IConnection connection; + + await _semaphore.WaitAsync(); + try + { + connection = _connection; + _connection = null; + _factory = null; + } + finally + { + _semaphore.Release(); + } + + RaiseConnectionReset(); + + if (connection != null) + { + try + { + await connection.DisposeAsync(); + } + catch (Exception ex) + { + Logging.LogException(ex); + } + } + } + public static async Task CreateConnection(string clientName) { if (_connection == null) diff --git a/Providers/Resgrid.Providers.Bus.Rabbit/RabbitInboundEventProvider.cs b/Providers/Resgrid.Providers.Bus.Rabbit/RabbitInboundEventProvider.cs index 1a5f4e5fc..26df73b1c 100644 --- a/Providers/Resgrid.Providers.Bus.Rabbit/RabbitInboundEventProvider.cs +++ b/Providers/Resgrid.Providers.Bus.Rabbit/RabbitInboundEventProvider.cs @@ -32,8 +32,51 @@ public class RabbitInboundEventProvider : IRabbitInboundEventProvider public async Task Start(string clientName, string queueName) { - await VerifyAndCreateClients(clientName); - await StartMonitoring(queueName); + // Dispose any channel from a previous Start (the host watchdog re-calls Start after a + // disconnect). Disposal also removes it from automatic-recovery tracking, so a late + // connection recovery can't resurrect the old consumer alongside the new one and + // double-deliver events. + await DisposeChannelAsync(); + + if (!await VerifyAndCreateClients(clientName)) + return; + + // _channel stays null when the connection couldn't be created; skip monitoring so the + // caller sees IsConnected() == false and can retry instead of an NRE killing the task. + if (_channel == null) + return; + + try + { + await StartMonitoring(queueName); + } + catch + { + // If consumer registration fails the channel is open but consumes nothing, so + // IsConnected() would report healthy and the host watchdog would never rebuild. + // Tear the channel down (nulled field makes IsConnected() false) and rethrow for + // the caller's retry path. + await DisposeChannelAsync(); + throw; + } + } + + private async Task DisposeChannelAsync() + { + var channel = _channel; + _channel = null; + + if (channel == null) + return; + + try + { + await channel.DisposeAsync(); + } + catch (Exception ex) + { + Logging.LogException(ex); + } } private async Task VerifyAndCreateClients(string clientName) @@ -55,6 +98,13 @@ private async Task VerifyAndCreateClients(string clientName) catch (Exception ex) { Framework.Logging.LogException(ex); + + // Exhausted channel numbers leave the connection open but unusable, and the host + // watchdog's retry would get the same connection back forever — reset it so the + // next Start builds a fresh one. + if (ex is RabbitMQ.Client.Exceptions.ChannelAllocationException) + await RabbitConnection.ForceResetAsync(); + return false; } diff --git a/Providers/Resgrid.Providers.Bus.Rabbit/RabbitInboundQueueProvider.cs b/Providers/Resgrid.Providers.Bus.Rabbit/RabbitInboundQueueProvider.cs index 6b4214ce6..6d62ad435 100644 --- a/Providers/Resgrid.Providers.Bus.Rabbit/RabbitInboundQueueProvider.cs +++ b/Providers/Resgrid.Providers.Bus.Rabbit/RabbitInboundQueueProvider.cs @@ -41,38 +41,69 @@ public RabbitInboundQueueProvider() public async Task Start(string clientName) { _clientName = clientName; + + // Dispose any channels from a previous Start (the watchdog re-calls Start after a + // disconnect). Disposal also removes them from automatic-recovery tracking, so a late + // connection recovery can't resurrect the old consumers alongside the new ones and + // double-process dispatches. + await DisposeChannelsAsync(); + var connection = await RabbitConnection.CreateConnection(clientName); if (connection != null) { - _channel = await connection.CreateChannelAsync(); - - if (CallQueueReceived != null) + try { - // Call dispatch has its own channel so no unrelated queue callback can delay an - // emergency notification, regardless of which backing store that callback uses. - _callChannel = await connection.CreateChannelAsync(); - await _callChannel.BasicQosAsync(0, 1, false); - } + _channel = await connection.CreateChannelAsync(); - if (UnitLocationEventQueueReceived != null) + if (CallQueueReceived != null) + { + // Call dispatch has its own channel so no unrelated queue callback can delay an + // emergency notification, regardless of which backing store that callback uses. + _callChannel = await connection.CreateChannelAsync(); + await _callChannel.BasicQosAsync(0, 1, false); + } + + if (UnitLocationEventQueueReceived != null) + { + _unitLocationChannel = await connection.CreateChannelAsync(); + var prefetchCount = (ushort)Math.Min( + ushort.MaxValue, + Math.Max(1, UnitTrackingConfig.UnitLocationQueuePrefetchCount)); + await _unitLocationChannel.BasicQosAsync(0, prefetchCount, false); + } + + if (PersonnelLocationEventQueueReceived != null) + { + // Personnel location storage must never serialize dispatch callbacks behind a slow + // Mongo/DocumentDB operation. Rabbit dispatches callbacks sequentially per channel. + _personnelLocationChannel = await connection.CreateChannelAsync(); + await _personnelLocationChannel.BasicQosAsync(0, 1, false); + } + + await StartMonitoring(); + } + catch (RabbitMQ.Client.Exceptions.ChannelAllocationException) { - _unitLocationChannel = await connection.CreateChannelAsync(); - var prefetchCount = (ushort)Math.Min( - ushort.MaxValue, - Math.Max(1, UnitTrackingConfig.UnitLocationQueuePrefetchCount)); - await _unitLocationChannel.BasicQosAsync(0, prefetchCount, false); + // The shared connection is open but out of channel numbers, so the watchdog's + // retry would get the same exhausted connection back from CreateConnection forever + // (the IsOpen guards can't see exhaustion). Dispose our channels first (clean close + // releases their numbers while the connection is still alive), then force-reset the + // connection so the retry builds a fresh one. + await DisposeChannelsAsync(); + await RabbitConnection.ForceResetAsync(); + throw; } - - if (PersonnelLocationEventQueueReceived != null) + catch { - // Personnel location storage must never serialize dispatch callbacks behind a slow - // Mongo/DocumentDB operation. Rabbit dispatches callbacks sequentially per channel. - _personnelLocationChannel = await connection.CreateChannelAsync(); - await _personnelLocationChannel.BasicQosAsync(0, 1, false); + // A partial startup must not linger: if StartMonitoring fails after the channels + // were created, every channel is open but consumers are incomplete, so IsConnected() + // would report healthy while nothing (or only some queues) is being consumed and the + // host watchdog would never rebuild. Tear everything down — nulled fields make + // IsConnected() false — and let the caller's retry path handle the failure. + await DisposeChannelsAsync(); + throw; } - - await StartMonitoring(); } } @@ -641,6 +672,30 @@ await _channel.BasicConsumeAsync( } } + private async Task DisposeChannelsAsync() + { + var channels = new[] { _channel, _callChannel, _unitLocationChannel, _personnelLocationChannel }; + _channel = null; + _callChannel = null; + _unitLocationChannel = null; + _personnelLocationChannel = null; + + foreach (var channel in channels) + { + if (channel == null) + continue; + + try + { + await channel.DisposeAsync(); + } + catch (Exception ex) + { + Logging.LogException(ex); + } + } + } + public bool IsConnected() { if (_channel == null || diff --git a/Providers/Resgrid.Providers.Bus.Rabbit/RabbitOutboundQueueProvider.cs b/Providers/Resgrid.Providers.Bus.Rabbit/RabbitOutboundQueueProvider.cs index f040dd833..07ba6f3c6 100644 --- a/Providers/Resgrid.Providers.Bus.Rabbit/RabbitOutboundQueueProvider.cs +++ b/Providers/Resgrid.Providers.Bus.Rabbit/RabbitOutboundQueueProvider.cs @@ -151,63 +151,25 @@ private async Task SendMessage(string queueName, string message, bool dura try { - var connection = await RabbitConnection.CreateConnection(_clientName); - if (connection != null) + return await PublishAsync(queueName, message, durable, expiration, requirePublisherConfirmation); + } + catch (RabbitMQ.Client.Exceptions.ChannelAllocationException ex) + { + // The shared connection still reports IsOpen when its channel numbers are exhausted, + // so the normal reconnect guards never fire and every send fails until the process + // restarts. Hard-reset the connection and retry the publish once on a fresh one. + Logging.LogException(ex); + + try { - // await using so the channel is closed via DisposeAsync(): the synchronous Dispose() on a - // v7 IChannel skips the async Channel.Close/CloseOk handshake that releases the channel - // number back to the SessionManager, leaking channels until the connection hits its limit - // (ChannelAllocationException: "The connection cannot support any more channels"). - var channelOptions = requirePublisherConfirmation - ? new CreateChannelOptions(true, true) - : null; - - await using (var channel = channelOptions == null - ? await connection.CreateChannelAsync() - : await connection.CreateChannelAsync(channelOptions)) - { - if (channel != null) - { - var props = new BasicProperties(); - props.Headers = new Dictionary(); - - if (durable) - { - props.DeliveryMode = DeliveryModes.Persistent; - props.Headers.Add("x-redelivered-count", 0); - } - else - props.DeliveryMode = DeliveryModes.Transient; - - props.Expiration = expiration; - - using var publishTimeout = requirePublisherConfirmation - ? new System.Threading.CancellationTokenSource( - TimeSpan.FromSeconds(Math.Max(1, UnitTrackingConfig.QueuePublishTimeoutSeconds))) - : null; - - await channel.BasicPublishAsync( - exchange: ServiceBusConfig.RabbbitExchange, - routingKey: RabbitConnection.SetQueueNameForEnv(queueName), - mandatory: true, - basicProperties: props, - body: Encoding.UTF8.GetBytes(message), - cancellationToken: publishTimeout?.Token ?? default); - - return true; - } - else - { - Logging.LogError("RabbitOutboundQueueProvider->SendMessage channel is null."); - } - } + await RabbitConnection.ForceResetAsync(); + return await PublishAsync(queueName, message, durable, expiration, requirePublisherConfirmation); } - else + catch (Exception retryEx) { - Logging.LogError("RabbitOutboundQueueProvider->SendMessage connection is null."); + Logging.LogException(retryEx); + return false; } - - return false; } catch (Exception ex) { @@ -216,6 +178,68 @@ await channel.BasicPublishAsync( } } + private async Task PublishAsync(string queueName, string message, bool durable, string expiration, + bool requirePublisherConfirmation) + { + var connection = await RabbitConnection.CreateConnection(_clientName); + if (connection != null) + { + // await using so the channel is closed via DisposeAsync(): the synchronous Dispose() on a + // v7 IChannel skips the async Channel.Close/CloseOk handshake that releases the channel + // number back to the SessionManager, leaking channels until the connection hits its limit + // (ChannelAllocationException: "The connection cannot support any more channels"). + var channelOptions = requirePublisherConfirmation + ? new CreateChannelOptions(true, true) + : null; + + await using (var channel = channelOptions == null + ? await connection.CreateChannelAsync() + : await connection.CreateChannelAsync(channelOptions)) + { + if (channel != null) + { + var props = new BasicProperties(); + props.Headers = new Dictionary(); + + if (durable) + { + props.DeliveryMode = DeliveryModes.Persistent; + props.Headers.Add("x-redelivered-count", 0); + } + else + props.DeliveryMode = DeliveryModes.Transient; + + props.Expiration = expiration; + + using var publishTimeout = requirePublisherConfirmation + ? new System.Threading.CancellationTokenSource( + TimeSpan.FromSeconds(Math.Max(1, UnitTrackingConfig.QueuePublishTimeoutSeconds))) + : null; + + await channel.BasicPublishAsync( + exchange: ServiceBusConfig.RabbbitExchange, + routingKey: RabbitConnection.SetQueueNameForEnv(queueName), + mandatory: true, + basicProperties: props, + body: Encoding.UTF8.GetBytes(message), + cancellationToken: publishTimeout?.Token ?? default); + + return true; + } + else + { + Logging.LogError("RabbitOutboundQueueProvider->SendMessage channel is null."); + } + } + } + else + { + Logging.LogError("RabbitOutboundQueueProvider->SendMessage connection is null."); + } + + return false; + } + private async Task SendMessagesWithConfirmation( string queueName, IReadOnlyCollection messages, @@ -231,42 +255,24 @@ private async Task SendMessagesWithConfirmation( try { - var connection = await RabbitConnection.CreateConnection(_clientName); - if (connection == null) - { - Logging.LogError("RabbitOutboundQueueProvider->SendMessagesWithConfirmation connection is null."); - return false; - } + return await PublishBatchAsync(queueName, messages, expiration, cancellationToken); + } + catch (RabbitMQ.Client.Exceptions.ChannelAllocationException ex) + { + // Same recovery as SendMessage: exhausted channel numbers leave the connection open + // but unusable, so reset it and retry the batch once. + Logging.LogException(ex); - await using var channel = - await connection.CreateChannelAsync(new CreateChannelOptions(true, true), cancellationToken); - var props = new BasicProperties + try { - DeliveryMode = DeliveryModes.Persistent, - Expiration = expiration, - Headers = new Dictionary - { - ["x-redelivered-count"] = 0 - } - }; - - using var publishTimeout = - CancellationTokenSource.CreateLinkedTokenSource(cancellationToken); - publishTimeout.CancelAfter( - TimeSpan.FromSeconds(Math.Max(1, UnitTrackingConfig.QueuePublishTimeoutSeconds))); - - foreach (var message in messages) + await RabbitConnection.ForceResetAsync(); + return await PublishBatchAsync(queueName, messages, expiration, cancellationToken); + } + catch (Exception retryEx) { - await channel.BasicPublishAsync( - exchange: ServiceBusConfig.RabbbitExchange, - routingKey: RabbitConnection.SetQueueNameForEnv(queueName), - mandatory: true, - basicProperties: props, - body: Encoding.UTF8.GetBytes(message), - cancellationToken: publishTimeout.Token); + Logging.LogException(retryEx); + return false; } - - return true; } catch (Exception ex) { @@ -275,6 +281,50 @@ await channel.BasicPublishAsync( } } + private async Task PublishBatchAsync( + string queueName, + IReadOnlyCollection messages, + string expiration, + CancellationToken cancellationToken) + { + var connection = await RabbitConnection.CreateConnection(_clientName); + if (connection == null) + { + Logging.LogError("RabbitOutboundQueueProvider->SendMessagesWithConfirmation connection is null."); + return false; + } + + await using var channel = + await connection.CreateChannelAsync(new CreateChannelOptions(true, true), cancellationToken); + var props = new BasicProperties + { + DeliveryMode = DeliveryModes.Persistent, + Expiration = expiration, + Headers = new Dictionary + { + ["x-redelivered-count"] = 0 + } + }; + + using var publishTimeout = + CancellationTokenSource.CreateLinkedTokenSource(cancellationToken); + publishTimeout.CancelAfter( + TimeSpan.FromSeconds(Math.Max(1, UnitTrackingConfig.QueuePublishTimeoutSeconds))); + + foreach (var message in messages) + { + await channel.BasicPublishAsync( + exchange: ServiceBusConfig.RabbbitExchange, + routingKey: RabbitConnection.SetQueueNameForEnv(queueName), + mandatory: true, + basicProperties: props, + body: Encoding.UTF8.GetBytes(message), + cancellationToken: publishTimeout.Token); + } + + return true; + } + public async Task VerifyAndCreateClients() { return await RabbitConnection.VerifyAndCreateClients(_clientName); diff --git a/Providers/Resgrid.Providers.Bus.Rabbit/RabbitTopicProvider.cs b/Providers/Resgrid.Providers.Bus.Rabbit/RabbitTopicProvider.cs index 123d6d2d7..c546ad30a 100644 --- a/Providers/Resgrid.Providers.Bus.Rabbit/RabbitTopicProvider.cs +++ b/Providers/Resgrid.Providers.Bus.Rabbit/RabbitTopicProvider.cs @@ -181,38 +181,28 @@ private async Task SendMessage(string topicName, string message, bool requ try { - var connection = await RabbitConnection.CreateConnection(_clientName); - if (connection == null) - return false; + return await PublishAsync(topicName, message, requirePublisherConfirmation); + } + catch (RabbitMQ.Client.Exceptions.ChannelAllocationException ex) + { + // The shared connection still reports IsOpen when its channel numbers are exhausted, + // so the normal reconnect guards never fire and every send fails until the process + // restarts. Hard-reset the connection and retry the publish once on a fresh one. + Framework.Logging.LogException(ex); - var channelOptions = requirePublisherConfirmation - ? new CreateChannelOptions(true, true) - : null; - await using (var channel = channelOptions == null - ? await connection.CreateChannelAsync() - : await connection.CreateChannelAsync(channelOptions)) + try { - using var publishTimeout = requirePublisherConfirmation - ? new System.Threading.CancellationTokenSource( - TimeSpan.FromSeconds(Math.Max(1, UnitTrackingConfig.QueuePublishTimeoutSeconds))) - : null; - await channel.BasicPublishAsync( - exchange: RabbitConnection.SetQueueNameForEnv(topicName), - routingKey: "", - mandatory: false, - basicProperties: new BasicProperties - { - DeliveryMode = requirePublisherConfirmation - ? DeliveryModes.Persistent - : DeliveryModes.Transient - }, - // UTF8: chat payloads carry emoji/unicode; superset of the ASCII previously used and - // the inbound consumer already decodes UTF8. - body: Encoding.UTF8.GetBytes(message), - cancellationToken: publishTimeout?.Token ?? default); - } + await RabbitConnection.ForceResetAsync(); - return true; + if (!await VerifyAndCreateClients(_clientName)) + return false; + + return await PublishAsync(topicName, message, requirePublisherConfirmation); + } + catch (Exception retryEx) + { + Framework.Logging.LogException(retryEx); + } } catch (Exception ex) { @@ -221,5 +211,41 @@ await channel.BasicPublishAsync( return false; } + + private async Task PublishAsync(string topicName, string message, bool requirePublisherConfirmation) + { + var connection = await RabbitConnection.CreateConnection(_clientName); + if (connection == null) + return false; + + var channelOptions = requirePublisherConfirmation + ? new CreateChannelOptions(true, true) + : null; + await using (var channel = channelOptions == null + ? await connection.CreateChannelAsync() + : await connection.CreateChannelAsync(channelOptions)) + { + using var publishTimeout = requirePublisherConfirmation + ? new System.Threading.CancellationTokenSource( + TimeSpan.FromSeconds(Math.Max(1, UnitTrackingConfig.QueuePublishTimeoutSeconds))) + : null; + await channel.BasicPublishAsync( + exchange: RabbitConnection.SetQueueNameForEnv(topicName), + routingKey: "", + mandatory: false, + basicProperties: new BasicProperties + { + DeliveryMode = requirePublisherConfirmation + ? DeliveryModes.Persistent + : DeliveryModes.Transient + }, + // UTF8: chat payloads carry emoji/unicode; superset of the ASCII previously used and + // the inbound consumer already decodes UTF8. + body: Encoding.UTF8.GetBytes(message), + cancellationToken: publishTimeout?.Token ?? default); + } + + return true; + } } } diff --git a/Tests/Resgrid.Tests/Bootstrapper.cs b/Tests/Resgrid.Tests/Bootstrapper.cs index d75214d01..f58e3c3a4 100644 --- a/Tests/Resgrid.Tests/Bootstrapper.cs +++ b/Tests/Resgrid.Tests/Bootstrapper.cs @@ -56,6 +56,16 @@ public static void Initialize() .As() .InstancePerLifetimeScope(); + // IncidentCommandService resolves chat services lazily through the ServiceLocator for + // its best-effort lane channel hooks. The real ChatChannelService can't activate in + // this container (its repository graph isn't registered), which logged an activation + // error on every lane save/delete test. Loose mocks turn the hooks into no-ops: + // un-setup async members return completed tasks with null results. + builder.RegisterInstance(new Moq.Mock().Object) + .As(); + builder.RegisterInstance(new Moq.Mock().Object) + .As(); + // UDF mock repositories builder.RegisterType() .As() diff --git a/Tests/Resgrid.Tests/Services/EncryptionServiceTests.cs b/Tests/Resgrid.Tests/Services/EncryptionServiceTests.cs index e2df91805..a0a2c1930 100644 --- a/Tests/Resgrid.Tests/Services/EncryptionServiceTests.cs +++ b/Tests/Resgrid.Tests/Services/EncryptionServiceTests.cs @@ -47,15 +47,41 @@ public void ShouldProduceDifferentCiphertextEachCallDueToRandomIv() } [Test] - public void ShouldProduceBase64Output() + public void ShouldProduceVersionedBase64Output() { var cipher = Sut.Encrypt("test"); + cipher.Should().StartWith("enc2:", "new ciphertexts carry the GCM format prefix"); + byte[] bytes = null; - Action act = () => { bytes = Convert.FromBase64String(cipher); }; + Action act = () => { bytes = Convert.FromBase64String(cipher.Substring("enc2:".Length)); }; act.Should().NotThrow(); bytes.Should().NotBeNull(); } + [Test] + public void ShouldThrowOnTamperedCiphertext() + { + var cipher = Sut.Encrypt("integrity matters"); + + // Flip one character in the Base64 body (past the prefix and nonce region). + var chars = cipher.ToCharArray(); + var index = chars.Length - 2; + chars[index] = chars[index] == 'A' ? 'B' : 'A'; + + Action act = () => Sut.Decrypt(new string(chars)); + act.Should().Throw("GCM authenticates the payload, so any tampering must fail the tag check"); + } + + [Test] + public void ShouldDecryptLegacyCbcCiphertext() + { + // Fixed pre-GCM (AES-256-CBC/PKCS7, IV-prefixed, unversioned Base64) ciphertext of + // "legacy global secret" under the fixture's test key/salt/iterations. Guards the + // legacy fallback path that existing data at rest depends on. + const string legacyCipher = "AQIDBAUGBwgJCgsMDQ4PEPTtkp0LPqxjlBC8ofdOfEjmKsxM3zYWcppjkR3460HA"; + Sut.Decrypt(legacyCipher).Should().Be("legacy global secret"); + } + [Test] public void ShouldThrowOnNullPlaintext() { @@ -121,12 +147,14 @@ public void DepartmentKeyShouldDifferFromGlobalKey() [Test] public void DifferentDepartmentsShouldProduceDifferentCiphertexts() { + // Deterministic with GCM: a wrong key always fails the authentication tag check. + // (Under the old CBC format this was a flaky padding-check assertion.) const string plainText = "shared secret"; Sut.EncryptForDepartment(plainText, 1, "DEPT1"); // ensures keys differ var cipher2 = Sut.EncryptForDepartment(plainText, 2, "DEPT2"); Action act = () => Sut.DecryptForDepartment(cipher2, 1, "DEPT1"); - act.Should().Throw("wrong department key should fail to decrypt"); + act.Should().Throw("wrong department key must fail the GCM tag check"); } [Test] @@ -136,7 +164,17 @@ public void SameDepartmentDifferentCodeShouldFailDecrypt() var cipher = Sut.EncryptForDepartment(plainText, 5, "ORIG"); Action act = () => Sut.DecryptForDepartment(cipher, 5, "DIFF"); - act.Should().Throw("changed department code produces a different key"); + act.Should().Throw("changed department code produces a different key, which must fail the GCM tag check"); + } + + [Test] + public void ShouldDecryptLegacyCbcDepartmentCiphertext() + { + // Fixed pre-GCM (AES-256-CBC/PKCS7, IV-prefixed, unversioned Base64) ciphertext of + // "legacy department secret" for department 5 / code "ORIG" under the fixture's test + // key/salt/iterations. Guards the legacy fallback for stored department credentials. + const string legacyCipher = "BwgJCgsMDQ4PEBESExQVFpXSN+nRnbOBO6F8HQ/JWEeucx3tCtsB9fO+TMX4ia4W"; + Sut.DecryptForDepartment(legacyCipher, 5, "ORIG").Should().Be("legacy department secret"); } [Test] diff --git a/Web/Resgrid.Web.Eventing/Worker.cs b/Web/Resgrid.Web.Eventing/Worker.cs index aea368520..e62d9fc2f 100644 --- a/Web/Resgrid.Web.Eventing/Worker.cs +++ b/Web/Resgrid.Web.Eventing/Worker.cs @@ -35,26 +35,78 @@ public Worker(IServiceProvider serviceProvider, IHubContext eventin _rabbitInboundEventProvider = scope.ServiceProvider.GetRequiredService(); } - protected override Task ExecuteAsync(CancellationToken stoppingToken = default) + protected override async Task ExecuteAsync(CancellationToken stoppingToken = default) { Console.WriteLine("Starting Eventing Worker"); stoppingToken.ThrowIfCancellationRequested(); - _rabbitInboundEventProvider.RegisterForEvents(PersonnelStatusUpdated, - UnitStatusUpdated, - CallsUpdated, - PersonnelStaffingUpdated, - CallAdded, - CallClosed, - PersonnelLocationUpdated, - UnitLocationUpdated, - IncidentCommandUpdated); + _rabbitInboundEventProvider.RegisterForEvents(PersonnelStatusUpdated, + UnitStatusUpdated, + CallsUpdated, + PersonnelStaffingUpdated, + CallAdded, + CallClosed, + PersonnelLocationUpdated, + UnitLocationUpdated, + IncidentCommandUpdated); + + _rabbitInboundEventProvider.RegisterForChatEvents(ChatEventReceived); + + await StartProviderAsync(); + + // Watchdog: the consumer channel dies silently if the shared Rabbit connection is ever + // replaced (e.g. RabbitConnection.ForceResetAsync after channel exhaustion, or a failed + // automatic recovery) — nothing restarts it and SignalR clients stop receiving updates + // until the pod is bounced. Rebuild the consumer after ~10s of continuous disconnect, + // retrying at most once a minute so a hard broker outage doesn't spin. + int disconnectedChecks = 0; + DateTime lastRestartAttemptUtc = DateTime.MinValue; + + while (!stoppingToken.IsCancellationRequested) + { + try + { + await Task.Delay(500, stoppingToken); + } + catch (OperationCanceledException) + { + break; + } + + if (_rabbitInboundEventProvider.IsConnected()) + { + disconnectedChecks = 0; + continue; + } + + disconnectedChecks++; - _rabbitInboundEventProvider.RegisterForChatEvents(ChatEventReceived); + if (disconnectedChecks >= 20 && (DateTime.UtcNow - lastRestartAttemptUtc) >= TimeSpan.FromSeconds(60)) + { + lastRestartAttemptUtc = DateTime.UtcNow; + + Console.WriteLine("Eventing Worker: Rabbit consumer disconnected; restarting event monitoring."); + await StartProviderAsync(); - _rabbitInboundEventProvider.Start("Eventing-Web", "EventingWeb").ConfigureAwait(false); + if (_rabbitInboundEventProvider.IsConnected()) + { + disconnectedChecks = 0; + Console.WriteLine("Eventing Worker: Event monitoring restarted."); + } + } + } + } - return Task.CompletedTask; + private async Task StartProviderAsync() + { + try + { + await _rabbitInboundEventProvider.Start("Eventing-Web", "EventingWeb"); + } + catch (Exception ex) + { + Resgrid.Framework.Logging.LogException(ex); + } } //public async Task StartAsync(CancellationToken cancellationToken = default) diff --git a/Web/Resgrid.Web.Services/Controllers/v4/ChatController.cs b/Web/Resgrid.Web.Services/Controllers/v4/ChatController.cs index 7d2d56e4d..bd395d80c 100644 --- a/Web/Resgrid.Web.Services/Controllers/v4/ChatController.cs +++ b/Web/Resgrid.Web.Services/Controllers/v4/ChatController.cs @@ -1257,7 +1257,7 @@ public async Task> GetPins(string channelId) [ProducesResponseType(StatusCodes.Status400BadRequest)] [ProducesResponseType(StatusCodes.Status401Unauthorized)] [ProducesResponseType(StatusCodes.Status404NotFound)] - public async Task> UploadAttachment(string channelId, string messageId, [FromForm] IFormFile file, CancellationToken cancellationToken) + public async Task> UploadAttachment(string channelId, string messageId, IFormFile file, CancellationToken cancellationToken) { if (!await ChatEnabledAsync()) return NotFound(); diff --git a/Web/Resgrid.Web/wwwroot/js/app/internal/dlist/resgrid.dlists.editlist.js b/Web/Resgrid.Web/wwwroot/js/app/internal/dlist/resgrid.dlists.editlist.js index 9e35039d6..2939a6847 100644 --- a/Web/Resgrid.Web/wwwroot/js/app/internal/dlist/resgrid.dlists.editlist.js +++ b/Web/Resgrid.Web/wwwroot/js/app/internal/dlist/resgrid.dlists.editlist.js @@ -35,21 +35,6 @@ var resgrid; } }); }); - function switchInputs(value) { - if (value) { - if (value === "Internal") { - $('#resgridEmail').show(); - $('#externalEmail').hide(); - } - else { - $('#resgridEmail').hide(); - $('#externalEmail').show(); - $('#emailError').hide(); - $("#submit_action").removeAttr("disabled"); - $('#List_EmailAddress').removeClass('input-validation-error'); - } - } - } function validateEmailAddress(emailAddress) { if (emailAddress) { if (!expression.test(emailAddress)) { @@ -78,7 +63,7 @@ var resgrid; }); } } - else if ($('#Type').select2('data').text === 'Internal') { + else if (typeof currentType !== 'undefined' && currentType === 'Internal') { $('#emailError').text('You need to specify an email address for the Internal type..'); $('#emailError').show(); $('#List_EmailAddress').addClass('input-validation-error'); diff --git a/Web/Resgrid.Web/wwwroot/js/app/internal/dlist/resgrid.dlists.newlist.js b/Web/Resgrid.Web/wwwroot/js/app/internal/dlist/resgrid.dlists.newlist.js index cb6f568c2..d7b427ec3 100644 --- a/Web/Resgrid.Web/wwwroot/js/app/internal/dlist/resgrid.dlists.newlist.js +++ b/Web/Resgrid.Web/wwwroot/js/app/internal/dlist/resgrid.dlists.newlist.js @@ -22,21 +22,6 @@ var resgrid; } }); }); - function switchInputs(value) { - if (value) { - if (value === "Internal") { - $('#resgridEmail').show(); - $('#externalEmail').hide(); - } - else { - $('#resgridEmail').hide(); - $('#externalEmail').show(); - $('#emailError').hide(); - $("#submit_action").removeAttr("disabled"); - $('#List_EmailAddress').removeClass('input-validation-error'); - } - } - } function validateEmailAddress(emailAddress) { if (emailAddress) { if (!expression.test(emailAddress)) { @@ -65,17 +50,13 @@ var resgrid; }); } } - else if ($('#Type').select2('data').text === 'Internal') { + else { + // New lists are always Internal (forced server-side), so a blank address is an error. $('#emailError').text('You need to specify an email address for the Internal type..'); $('#emailError').show(); $('#List_EmailAddress').addClass('input-validation-error'); $('#submit_action').attr("disabled", "disabled"); } - else { - $('#emailError').hide(); - $("#submit_action").removeAttr("disabled"); - $('#List_EmailAddress').removeClass('input-validation-error'); - } } newlist.validateEmailAddress = validateEmailAddress; var expression = /^(([^<>()[\]\\.,;:\s@\"]+(\.[^<>()[\]\\.,;:\s@\"]+)*)|(\".+\"))$/; diff --git a/Web/Resgrid.Web/wwwroot/js/app/internal/resgrid.user.js b/Web/Resgrid.Web/wwwroot/js/app/internal/resgrid.user.js index dc88d6884..c61a6fcbb 100644 --- a/Web/Resgrid.Web/wwwroot/js/app/internal/resgrid.user.js +++ b/Web/Resgrid.Web/wwwroot/js/app/internal/resgrid.user.js @@ -8,6 +8,16 @@ var resgrid; (function (resgrid) { var main; (function (main) { + // jQuery Validation 1.19.0 crashes on blur of contenteditable elements that have no name + // attribute (Quill editors: errorsFor -> escapeCssMeta(undefined) -> TypeError). Keep + // contenteditable out of validation entirely; Quill content is posted via hidden inputs. + // Runs before document ready, so unobtrusive validation picks the default up when it parses. + if ($.validator) { + $.validator.setDefaults({ + ignore: ':hidden, [contenteditable]' + }); + } + // Expired auth cookies leave background polls (DataTables ajax, etc.) failing with 401s; // the ajaxError handler below redirects to login, so only the 401 alert is silenced here. // Every other DataTables error keeps the library's default alert so real problems surface. diff --git a/Workers/Resgrid.Workers.Console/Tasks/QueuesProcessorTask.cs b/Workers/Resgrid.Workers.Console/Tasks/QueuesProcessorTask.cs index f93acc865..2e832a08a 100644 --- a/Workers/Resgrid.Workers.Console/Tasks/QueuesProcessorTask.cs +++ b/Workers/Resgrid.Workers.Console/Tasks/QueuesProcessorTask.cs @@ -53,11 +53,58 @@ public async Task ProcessAsync(QueuesProcessorCommand command, IQuidjiboProgress queue.WorkflowQueueReceived += OnWorkflowQueueReceived; queue.ChatbotMessageQueueReceived += OnChatbotMessageReceived; - await queue.Start("QueueProcessor-CQRS"); + try + { + await queue.Start("QueueProcessor-CQRS"); + } + catch (Exception ex) + { + // A failed initial start (broker briefly down at boot, partial startup torn down by + // the provider) must not kill this task before the watchdog loop below exists — + // the loop sees IsConnected() == false and retries. + Resgrid.Framework.Logging.LogException(ex); + } + + // Watchdog: the consumer channels die silently if the shared Rabbit connection is ever + // replaced (e.g. RabbitConnection.ForceResetAsync after channel exhaustion, or a failed + // automatic recovery) — nothing restarts them and dispatch processing stops until the pod + // is bounced. Rebuild the consumers after ~10s of continuous disconnect, retrying at most + // once a minute so a hard broker outage doesn't spin. + int disconnectedChecks = 0; + DateTime lastRestartAttemptUtc = DateTime.MinValue; while (!_cancellationToken.IsCancellationRequested) { Thread.Sleep(500); + + if (queue.IsConnected()) + { + disconnectedChecks = 0; + continue; + } + + disconnectedChecks++; + + if (disconnectedChecks >= 20 && (DateTime.UtcNow - lastRestartAttemptUtc) >= TimeSpan.FromSeconds(60)) + { + lastRestartAttemptUtc = DateTime.UtcNow; + + try + { + _logger.LogWarning($"{Name}: Queue consumers disconnected; restarting queue monitoring."); + await queue.Start("QueueProcessor-CQRS"); + + if (queue.IsConnected()) + { + disconnectedChecks = 0; + _logger.LogInformation($"{Name}: Queue monitoring restarted."); + } + } + catch (Exception ex) + { + Resgrid.Framework.Logging.LogException(ex); + } + } } if (progress != null)