diff --git a/dev/.env.example b/dev/.env.example index d0ebf50efb..f0aed83a59 100644 --- a/dev/.env.example +++ b/dev/.env.example @@ -20,4 +20,8 @@ IDP_SP_ACS_URL=http://localhost:51822/saml2/yourOrgIdHere/Acs # Optional reverse proxy configuration # Should match server listen ports in reverse-proxy.conf API_PROXY_PORT=4100 -IDENTITY_PROXY_PORT=33756 \ No newline at end of file +IDENTITY_PROXY_PORT=33756 + +# Optional RabbitMQ configuration +RABBITMQ_DEFAULT_USER=bitwarden +RABBITMQ_DEFAULT_PASS=SET_A_PASSWORD_HERE_123 diff --git a/dev/docker-compose.yml b/dev/docker-compose.yml index c02d3c872b..d23eaefbb0 100644 --- a/dev/docker-compose.yml +++ b/dev/docker-compose.yml @@ -84,6 +84,20 @@ services: profiles: - idp + rabbitmq: + image: rabbitmq:management + container_name: rabbitmq + ports: + - "5672:5672" + - "15672:15672" + environment: + RABBITMQ_DEFAULT_USER: ${RABBITMQ_DEFAULT_USER} + RABBITMQ_DEFAULT_PASS: ${RABBITMQ_DEFAULT_PASS} + volumes: + - rabbitmq_data:/var/lib/rabbitmq_data + profiles: + - rabbitmq + reverse-proxy: image: nginx:alpine container_name: reverse-proxy @@ -99,3 +113,4 @@ volumes: mssql_dev_data: postgres_dev_data: mysql_dev_data: + rabbitmq_data: diff --git a/src/Core/AdminConsole/Services/Implementations/RabbitMqEventHttpPostListener.cs b/src/Core/AdminConsole/Services/Implementations/RabbitMqEventHttpPostListener.cs new file mode 100644 index 0000000000..5a875f9278 --- /dev/null +++ b/src/Core/AdminConsole/Services/Implementations/RabbitMqEventHttpPostListener.cs @@ -0,0 +1,35 @@ +using System.Net.Http.Json; +using Bit.Core.Models.Data; +using Bit.Core.Settings; +using Microsoft.Extensions.Logging; + +namespace Bit.Core.Services; + +public class RabbitMqEventHttpPostListener : RabbitMqEventListenerBase +{ + private readonly HttpClient _httpClient; + private readonly string _httpPostUrl; + private readonly string _queueName; + + protected override string QueueName => _queueName; + + public const string HttpClientName = "EventHttpPostListenerHttpClient"; + + public RabbitMqEventHttpPostListener( + IHttpClientFactory httpClientFactory, + ILogger logger, + GlobalSettings globalSettings) + : base(logger, globalSettings) + { + _httpClient = httpClientFactory.CreateClient(HttpClientName); + _httpPostUrl = globalSettings.EventLogging.RabbitMq.HttpPostUrl; + _queueName = globalSettings.EventLogging.RabbitMq.HttpPostQueueName; + } + + protected override async Task HandleMessageAsync(EventMessage eventMessage) + { + var content = JsonContent.Create(eventMessage); + var response = await _httpClient.PostAsync(_httpPostUrl, content); + response.EnsureSuccessStatusCode(); + } +} diff --git a/src/Core/AdminConsole/Services/Implementations/RabbitMqEventListenerBase.cs b/src/Core/AdminConsole/Services/Implementations/RabbitMqEventListenerBase.cs new file mode 100644 index 0000000000..48a549d261 --- /dev/null +++ b/src/Core/AdminConsole/Services/Implementations/RabbitMqEventListenerBase.cs @@ -0,0 +1,93 @@ +using System.Text.Json; +using Bit.Core.Models.Data; +using Bit.Core.Settings; +using Microsoft.Extensions.Hosting; +using Microsoft.Extensions.Logging; +using RabbitMQ.Client; +using RabbitMQ.Client.Events; + +namespace Bit.Core.Services; + +public abstract class RabbitMqEventListenerBase : BackgroundService +{ + private IChannel _channel; + private IConnection _connection; + private readonly string _exchangeName; + private readonly ConnectionFactory _factory; + private readonly ILogger _logger; + + protected abstract string QueueName { get; } + + protected RabbitMqEventListenerBase( + ILogger logger, + GlobalSettings globalSettings) + { + _factory = new ConnectionFactory + { + HostName = globalSettings.EventLogging.RabbitMq.HostName, + UserName = globalSettings.EventLogging.RabbitMq.Username, + Password = globalSettings.EventLogging.RabbitMq.Password + }; + _exchangeName = globalSettings.EventLogging.RabbitMq.ExchangeName; + _logger = logger; + } + + public override async Task StartAsync(CancellationToken cancellationToken) + { + _connection = await _factory.CreateConnectionAsync(cancellationToken); + _channel = await _connection.CreateChannelAsync(cancellationToken: cancellationToken); + + await _channel.ExchangeDeclareAsync(exchange: _exchangeName, type: ExchangeType.Fanout, durable: true); + await _channel.QueueDeclareAsync(queue: QueueName, + durable: true, + exclusive: false, + autoDelete: false, + arguments: null, + cancellationToken: cancellationToken); + await _channel.QueueBindAsync(queue: QueueName, + exchange: _exchangeName, + routingKey: string.Empty, + cancellationToken: cancellationToken); + await base.StartAsync(cancellationToken); + } + + protected override async Task ExecuteAsync(CancellationToken stoppingToken) + { + var consumer = new AsyncEventingBasicConsumer(_channel); + consumer.ReceivedAsync += async (_, eventArgs) => + { + try + { + var eventMessage = JsonSerializer.Deserialize(eventArgs.Body.Span); + await HandleMessageAsync(eventMessage); + } + catch (Exception ex) + { + _logger.LogError(ex, "An error occurred while processing the message"); + } + }; + + await _channel.BasicConsumeAsync(QueueName, autoAck: true, consumer: consumer, cancellationToken: stoppingToken); + + while (!stoppingToken.IsCancellationRequested) + { + await Task.Delay(1_000, stoppingToken); + } + } + + public override async Task StopAsync(CancellationToken cancellationToken) + { + await _channel.CloseAsync(); + await _connection.CloseAsync(); + await base.StopAsync(cancellationToken); + } + + public override void Dispose() + { + _channel.Dispose(); + _connection.Dispose(); + base.Dispose(); + } + + protected abstract Task HandleMessageAsync(EventMessage eventMessage); +} diff --git a/src/Core/AdminConsole/Services/Implementations/RabbitMqEventRepositoryListener.cs b/src/Core/AdminConsole/Services/Implementations/RabbitMqEventRepositoryListener.cs new file mode 100644 index 0000000000..25d85bddeb --- /dev/null +++ b/src/Core/AdminConsole/Services/Implementations/RabbitMqEventRepositoryListener.cs @@ -0,0 +1,29 @@ +using Bit.Core.Models.Data; +using Bit.Core.Settings; +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Logging; + +namespace Bit.Core.Services; + +public class RabbitMqEventRepositoryListener : RabbitMqEventListenerBase +{ + private readonly IEventWriteService _eventWriteService; + private readonly string _queueName; + + protected override string QueueName => _queueName; + + public RabbitMqEventRepositoryListener( + [FromKeyedServices("persistent")] IEventWriteService eventWriteService, + ILogger logger, + GlobalSettings globalSettings) + : base(logger, globalSettings) + { + _eventWriteService = eventWriteService; + _queueName = globalSettings.EventLogging.RabbitMq.EventRepositoryQueueName; + } + + protected override Task HandleMessageAsync(EventMessage eventMessage) + { + return _eventWriteService.CreateAsync(eventMessage); + } +} diff --git a/src/Core/AdminConsole/Services/Implementations/RabbitMqEventWriteService.cs b/src/Core/AdminConsole/Services/Implementations/RabbitMqEventWriteService.cs new file mode 100644 index 0000000000..d89cf890ac --- /dev/null +++ b/src/Core/AdminConsole/Services/Implementations/RabbitMqEventWriteService.cs @@ -0,0 +1,65 @@ +using System.Text.Json; +using Bit.Core.Models.Data; +using Bit.Core.Settings; +using RabbitMQ.Client; + +namespace Bit.Core.Services; +public class RabbitMqEventWriteService : IEventWriteService, IAsyncDisposable +{ + private readonly ConnectionFactory _factory; + private readonly Lazy> _lazyConnection; + private readonly string _exchangeName; + + public RabbitMqEventWriteService(GlobalSettings globalSettings) + { + _factory = new ConnectionFactory + { + HostName = globalSettings.EventLogging.RabbitMq.HostName, + UserName = globalSettings.EventLogging.RabbitMq.Username, + Password = globalSettings.EventLogging.RabbitMq.Password + }; + _exchangeName = globalSettings.EventLogging.RabbitMq.ExchangeName; + + _lazyConnection = new Lazy>(CreateConnectionAsync); + } + + public async Task CreateAsync(IEvent e) + { + var connection = await _lazyConnection.Value; + using var channel = await connection.CreateChannelAsync(); + + await channel.ExchangeDeclareAsync(exchange: _exchangeName, type: ExchangeType.Fanout, durable: true); + + var body = JsonSerializer.SerializeToUtf8Bytes(e); + + await channel.BasicPublishAsync(exchange: _exchangeName, routingKey: string.Empty, body: body); + } + + public async Task CreateManyAsync(IEnumerable events) + { + var connection = await _lazyConnection.Value; + using var channel = await connection.CreateChannelAsync(); + await channel.ExchangeDeclareAsync(exchange: _exchangeName, type: ExchangeType.Fanout, durable: true); + + foreach (var e in events) + { + var body = JsonSerializer.SerializeToUtf8Bytes(e); + + await channel.BasicPublishAsync(exchange: _exchangeName, routingKey: string.Empty, body: body); + } + } + + public async ValueTask DisposeAsync() + { + if (_lazyConnection.IsValueCreated) + { + var connection = await _lazyConnection.Value; + await connection.DisposeAsync(); + } + } + + private async Task CreateConnectionAsync() + { + return await _factory.CreateConnectionAsync(); + } +} diff --git a/src/Core/Core.csproj b/src/Core/Core.csproj index 7a5f7e2543..210a33f3f7 100644 --- a/src/Core/Core.csproj +++ b/src/Core/Core.csproj @@ -40,7 +40,7 @@ - + @@ -70,12 +70,13 @@ + - + diff --git a/src/Core/Settings/GlobalSettings.cs b/src/Core/Settings/GlobalSettings.cs index 97d66aed53..718293891b 100644 --- a/src/Core/Settings/GlobalSettings.cs +++ b/src/Core/Settings/GlobalSettings.cs @@ -53,6 +53,7 @@ public class GlobalSettings : IGlobalSettings public virtual SqlSettings PostgreSql { get; set; } = new SqlSettings(); public virtual SqlSettings MySql { get; set; } = new SqlSettings(); public virtual SqlSettings Sqlite { get; set; } = new SqlSettings() { ConnectionString = "Data Source=:memory:" }; + public virtual EventLoggingSettings EventLogging { get; set; } = new EventLoggingSettings(); public virtual MailSettings Mail { get; set; } = new MailSettings(); public virtual IConnectionStringSettings Storage { get; set; } = new ConnectionStringSettings(); public virtual ConnectionStringSettings Events { get; set; } = new ConnectionStringSettings(); @@ -256,6 +257,44 @@ public class GlobalSettings : IGlobalSettings } } + public class EventLoggingSettings + { + public RabbitMqSettings RabbitMq { get; set; } + + public class RabbitMqSettings + { + private string _hostName; + private string _username; + private string _password; + private string _exchangeName; + + public virtual string EventRepositoryQueueName { get; set; } = "events-write-queue"; + public virtual string HttpPostQueueName { get; set; } = "events-httpPost-queue"; + public virtual string HttpPostUrl { get; set; } + + public string HostName + { + get => _hostName; + set => _hostName = value.Trim('"'); + } + public string Username + { + get => _username; + set => _username = value.Trim('"'); + } + public string Password + { + get => _password; + set => _password = value.Trim('"'); + } + public string ExchangeName + { + get => _exchangeName; + set => _exchangeName = value.Trim('"'); + } + } + } + public class ConnectionStringSettings : IConnectionStringSettings { private string _connectionString; diff --git a/src/Events/Startup.cs b/src/Events/Startup.cs index bac39c68dd..03e99f14e8 100644 --- a/src/Events/Startup.cs +++ b/src/Events/Startup.cs @@ -82,6 +82,22 @@ public class Startup { services.AddHostedService(); } + + // Optional RabbitMQ Listeners + if (CoreHelpers.SettingHasValue(globalSettings.EventLogging.RabbitMq.HostName) && + CoreHelpers.SettingHasValue(globalSettings.EventLogging.RabbitMq.Username) && + CoreHelpers.SettingHasValue(globalSettings.EventLogging.RabbitMq.Password) && + CoreHelpers.SettingHasValue(globalSettings.EventLogging.RabbitMq.ExchangeName)) + { + services.AddKeyedSingleton("persistent"); + services.AddHostedService(); + + if (CoreHelpers.SettingHasValue(globalSettings.EventLogging.RabbitMq.HttpPostUrl)) + { + services.AddHttpClient(RabbitMqEventHttpPostListener.HttpClientName); + services.AddHostedService(); + } + } } public void Configure( diff --git a/src/SharedWeb/Utilities/ServiceCollectionExtensions.cs b/src/SharedWeb/Utilities/ServiceCollectionExtensions.cs index e1369d5366..622b3d7f39 100644 --- a/src/SharedWeb/Utilities/ServiceCollectionExtensions.cs +++ b/src/SharedWeb/Utilities/ServiceCollectionExtensions.cs @@ -325,7 +325,17 @@ public static class ServiceCollectionExtensions } else if (globalSettings.SelfHosted) { - services.AddSingleton(); + if (CoreHelpers.SettingHasValue(globalSettings.EventLogging.RabbitMq.HostName) && + CoreHelpers.SettingHasValue(globalSettings.EventLogging.RabbitMq.Username) && + CoreHelpers.SettingHasValue(globalSettings.EventLogging.RabbitMq.Password) && + CoreHelpers.SettingHasValue(globalSettings.EventLogging.RabbitMq.ExchangeName)) + { + services.AddSingleton(); + } + else + { + services.AddSingleton(); + } } else {