From e2ac5d117a960f1d8e282df8eaa3fb80f35cebb5 Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Sat, 15 Aug 2026 12:17:22 -0400 Subject: [PATCH] perf(audit): bounded async audit writer with batched inserts, one-time bootstrap, retention sweep --- docs/DesignDecisions.md | 25 ++ docs/GatewayConfiguration.md | 1 + .../Configuration/GatewayOptionsValidator.cs | 7 + .../Configuration/SecurityOptions.cs | 9 + .../Security/Audit/CanonicalAuditWriter.cs | 20 +- .../Security/Audit/ChannelAuditWriter.cs | 333 ++++++++++++++++++ .../Audit/SqliteCanonicalAuditStore.cs | 189 ++++++++-- .../AuthStoreServiceCollectionExtensions.cs | 28 +- .../Security/Audit/ChannelAuditWriterTests.cs | 271 ++++++++++++++ 9 files changed, 845 insertions(+), 38 deletions(-) create mode 100644 src/ZB.MOM.WW.MxGateway.Server/Security/Audit/ChannelAuditWriter.cs create mode 100644 src/ZB.MOM.WW.MxGateway.Tests/Security/Audit/ChannelAuditWriterTests.cs diff --git a/docs/DesignDecisions.md b/docs/DesignDecisions.md index fcaeafd..2d1e479 100644 --- a/docs/DesignDecisions.md +++ b/docs/DesignDecisions.md @@ -228,6 +228,31 @@ Storage recommendation: administrators. - Require TLS when the gateway is reachable off-machine. +## Audit Pipeline + +Decision: audit is asynchronous, bounded, and swept. + +The canonical `IAuditWriter` contract has always been best-effort — a failed audit write is +logged and swallowed so it cannot abort the action that produced it. The registered writer is +`ChannelAuditWriter`, which makes the cost of that promise explicit: a producer enqueues onto a +4096-event bounded channel and returns, and `AuditDrainService` commits up to 64 buffered events +per transaction. This exists because constraint denials are emitted per denied tag inside bulk +RPC loops: a partially denied 1,000-tag request previously awaited 1,000 sequential SQLite +inserts — each re-running `CREATE TABLE IF NOT EXISTS` — against the same database file every +authenticated call reads. The schema bootstrap now runs once, from the drain's `StartAsync`. + +When the channel is full the newest event is dropped and counted rather than blocking the +producer: a stalled audit database must cost audit completeness, not gateway availability. Drops +are logged once and reported in aggregate on each sweep. Shutdown drains what is buffered under a +2-second cap. Where no hosted service runs — the `apikey` admin CLI — the writer falls back to the +synchronous path, so audit is never buffered into a channel nobody drains. + +`MxGateway:Security:AuditRetentionDays` (default 90, minimum 1) bounds the table: the drain sweeps +at startup and hourly, deleting older rows. Retention cannot be configured off. The sweep compares +through SQLite's `datetime()` rather than on the stored ISO-8601 text, because timestamps written +from a non-UTC offset do not sort lexicographically against a UTC cutoff; a row whose timestamp +cannot be parsed yields NULL and is kept rather than deleted. + ## Authorization Decision: start with scope checks by command category. diff --git a/docs/GatewayConfiguration.md b/docs/GatewayConfiguration.md index 23bf2a4..9d269e1 100644 --- a/docs/GatewayConfiguration.md +++ b/docs/GatewayConfiguration.md @@ -393,6 +393,7 @@ model requires otherwise. | `MxGateway:Security:ApiKeyFailureWindowSeconds` | `60` | Sliding-window length, in seconds, over which API-key verification failures are counted, for both the per-partition and the per-key-id aggregate layer. Must be greater than zero. | | `MxGateway:Security:ApiKeyFailureAggregateLimit` | `30` | Failed verifications for one key id counted across **all** transport peers within `ApiKeyFailureWindowSeconds` before that key id enters probe mode. This second layer bounds a distributed or source-rotating sprayer that never trips any single `(peer, key id)` partition. `0` disables the aggregate layer, leaving only per-partition counting. Must be zero or greater. | | `MxGateway:Security:ApiKeyFailureProbeIntervalSeconds` | `5` | Minimum interval, in seconds, between probe admissions for an over-limit partition or key-id aggregate. An over-limit state is a valve rather than a wall: one request per interval reaches the real verifier — exactly one, even when a burst arrives together at the interval boundary — so the holder of the correct secret always gets through and clears the state, while everything else is still refused before the store read. `0` blocks absolutely instead — **not recommended**, because an unauthenticated peer can then deny the key to its holder for the whole window. Must be zero or greater. | +| `MxGateway:Security:AuditRetentionDays` | `90` | Days of canonical audit history kept in the `audit_event` table. The audit drain sweeps once at startup and hourly thereafter, deleting rows older than this window; without it the table grows without bound inside the same SQLite file the authentication hot path reads. Rows whose timestamp SQLite cannot parse are never swept. Must be greater than zero — retention can be widened but not switched off. | | `MxGateway:Security:ApiKeyFailureTrackedPeers` | `4096` | Maximum distinct partitions tracked by the failure counter (a bounded LRU) so a spray of unique tokens cannot grow memory without limit. It cannot be used to flush an active block either: only a validly shaped `mxgw__` token mints a key-id partition (everything else lands on the sender's transport-peer partition), each address may mint at most 32 key-id partitions before the overflow collapses onto that address's fallback partition, and eviction prefers fully expired windows, never removing an over-limit partition until the map exceeds twice this cap. Must be greater than zero. | ## Galaxy Options diff --git a/src/ZB.MOM.WW.MxGateway.Server/Configuration/GatewayOptionsValidator.cs b/src/ZB.MOM.WW.MxGateway.Server/Configuration/GatewayOptionsValidator.cs index 3732209..3f98691 100644 --- a/src/ZB.MOM.WW.MxGateway.Server/Configuration/GatewayOptionsValidator.cs +++ b/src/ZB.MOM.WW.MxGateway.Server/Configuration/GatewayOptionsValidator.cs @@ -100,6 +100,13 @@ public sealed class GatewayOptionsValidator : OptionsValidatorBase public int ApiKeyFailureTrackedPeers { get; init; } = 4096; + + /// + /// Gets how many days of canonical audit history the gateway keeps. The audit drain sweeps + /// audit_event once at startup and hourly thereafter, deleting rows older than this + /// window; without it the table grows without bound in the same SQLite file the + /// authentication hot path reads. Must be greater than zero — audit retention cannot be + /// disabled by configuration, only widened. Default is 90 days. + /// + public int AuditRetentionDays { get; init; } = 90; } diff --git a/src/ZB.MOM.WW.MxGateway.Server/Security/Audit/CanonicalAuditWriter.cs b/src/ZB.MOM.WW.MxGateway.Server/Security/Audit/CanonicalAuditWriter.cs index 593849b..a03fdfb 100644 --- a/src/ZB.MOM.WW.MxGateway.Server/Security/Audit/CanonicalAuditWriter.cs +++ b/src/ZB.MOM.WW.MxGateway.Server/Security/Audit/CanonicalAuditWriter.cs @@ -3,24 +3,26 @@ using ZB.MOM.WW.Audit; namespace ZB.MOM.WW.MxGateway.Server.Security.Audit; /// -/// Best-effort over the MxGateway-owned -/// . It honours the canonical +/// Best-effort, synchronous over the MxGateway-owned +/// . It honours the canonical /// contract: a failed audit write is swallowed and logged /// rather than propagated, so it can never abort the user-facing action that produced it. /// /// -/// This is the single sink through which ALL MxGateway audit flows — the library admin -/// verbs (via ) and the gateway's own -/// dashboard / constraint-denial producers, which write canonical events directly. The -/// best-effort wrapping here also closes the gap that the library's +/// This is the durable bottom of the audit pipeline. Callers reach it two ways: through +/// — the registered , which +/// enqueues and lets batch events onto the sink — and +/// directly, when there is no drain to batch behind (the apikey CLI, and any host +/// shutdown window), where writing through immediately is the only way the event survives. +/// The best-effort wrapping here also closes the gap that the library's /// SqliteApiKeyAuditStore.AppendAsync propagated exceptions. /// public sealed class CanonicalAuditWriter( - SqliteCanonicalAuditStore store, + IAuditEventSink sink, ILogger logger) : IAuditWriter { /// - /// Persists a canonical audit event to the underlying . + /// Persists a canonical audit event to the underlying . /// Any failure is caught, logged, and swallowed rather than propagated to the caller. /// /// The canonical audit event to persist. @@ -32,7 +34,7 @@ public sealed class CanonicalAuditWriter( try { - await store.InsertAsync(auditEvent, cancellationToken).ConfigureAwait(false); + await sink.InsertAsync(auditEvent, cancellationToken).ConfigureAwait(false); } catch (Exception exception) { diff --git a/src/ZB.MOM.WW.MxGateway.Server/Security/Audit/ChannelAuditWriter.cs b/src/ZB.MOM.WW.MxGateway.Server/Security/Audit/ChannelAuditWriter.cs new file mode 100644 index 0000000..3b204cf --- /dev/null +++ b/src/ZB.MOM.WW.MxGateway.Server/Security/Audit/ChannelAuditWriter.cs @@ -0,0 +1,333 @@ +using System.Threading.Channels; +using ZB.MOM.WW.Audit; +using ZB.MOM.WW.MxGateway.Server.Configuration; + +namespace ZB.MOM.WW.MxGateway.Server.Security.Audit; + +/// +/// Bounded, non-blocking : enqueues onto a +/// fixed-capacity channel and returns, leaving to batch the +/// events onto the durable . +/// +/// +/// The canonical contract is already best-effort — a failed audit +/// write is swallowed rather than propagated. The channel makes the bound on that +/// promise explicit: audit can cost the calling RPC at most one enqueue, never a SQLite +/// round-trip, and at most events of memory. This matters on the +/// constraint-denial path, where a partially denied bulk RPC previously awaited one insert per +/// denied tag, serially, against the same database file the authentication hot path reads. +/// +/// When the channel is full the newest write is dropped () +/// and counted in . Dropping is the deliberate choice over blocking: +/// a stalled audit database must degrade audit completeness, not stall the gateway. Drops are +/// logged, and reports the running total on its sweep. +/// +/// +/// Until a drain attaches (), and again after it detaches, writes go +/// straight through to . Enqueueing into a channel nobody will +/// ever read would silently discard audit in the processes that have no hosted services — the +/// apikey admin CLI and the DI-only tests — so those keep the original synchronous path. +/// +/// +public sealed class ChannelAuditWriter : IAuditWriter +{ + /// + /// Maximum number of audit events buffered before writes start being dropped. Sized to + /// absorb a fully denied bulk RPC (the gateway's bulk request cap) plus headroom, so a + /// realistic burst is buffered rather than lost. + /// + public const int ChannelCapacity = 4096; + + private readonly CanonicalAuditWriter _directWriter; + private readonly ILogger _logger; + private readonly Channel _channel; + + private long _droppedCount; + private int _drainAttached; + private int _dropLogged; + + /// Creates the writer and its bounded buffer. + /// The synchronous writer used when no drain is attached. + /// Logger for drop diagnostics. + public ChannelAuditWriter(CanonicalAuditWriter directWriter, ILogger logger) + { + _directWriter = directWriter; + _logger = logger; + + // DropWrite discards the incoming item and still reports success to the producer, so the + // itemDropped callback is the only place a drop can be observed and counted. + _channel = Channel.CreateBounded( + new BoundedChannelOptions(ChannelCapacity) + { + FullMode = BoundedChannelFullMode.DropWrite, + SingleReader = true, + SingleWriter = false, + }, + itemDropped: RecordDrop); + } + + /// Gets the number of audit events dropped because the channel was full. + public long DroppedCount => Interlocked.Read(ref _droppedCount); + + /// Gets the reader the drain service consumes buffered events from. + public ChannelReader Reader => _channel.Reader; + + /// + /// Marks a drain as running, so subsequent writes enqueue instead of writing through. + /// Called by once its one-time bootstrap has completed. + /// + public void AttachDrain() => Volatile.Write(ref _drainAttached, 1); + + /// + /// Marks the drain as no longer running, so writes revert to the synchronous path. Called + /// at shutdown so late audit is still persisted rather than buffered into a dead channel. + /// + public void DetachDrain() => Volatile.Write(ref _drainAttached, 0); + + /// + /// Enqueues a canonical audit event for the drain to persist. Never blocks, never throws, + /// and never touches the store on the caller's thread while a drain is attached. + /// + /// The canonical audit event to persist. + /// Token honoured only by the direct write-through path. + /// A task that represents the asynchronous operation. + public Task WriteAsync(AuditEvent auditEvent, CancellationToken cancellationToken = default) + { + ArgumentNullException.ThrowIfNull(auditEvent); + + if (Volatile.Read(ref _drainAttached) == 0) + { + return _directWriter.WriteAsync(auditEvent, cancellationToken); + } + + // Under DropWrite a full channel still reports success (the discard surfaces through the + // itemDropped callback); a false here means the channel is completed, which is also a drop. + if (!_channel.Writer.TryWrite(auditEvent)) + { + RecordDrop(auditEvent); + } + + return Task.CompletedTask; + } + + /// Signals that no further events will be enqueued, so the drain loop can finish. + public void CompleteWriting() => _channel.Writer.TryComplete(); + + private void RecordDrop(AuditEvent auditEvent) + { + Interlocked.Increment(ref _droppedCount); + + // Log the first drop only; the running total is reported on the drain's periodic sweep, + // so a sustained overload cannot turn audit pressure into a log flood. + if (Interlocked.Exchange(ref _dropLogged, 1) == 0) + { + _logger.LogWarning( + "Audit channel is full ({Capacity} events); dropping audit event {EventId} (action {Action}). " + + "Audit is best-effort and bounded; further drops are reported in aggregate.", + ChannelCapacity, + auditEvent.EventId, + auditEvent.Action); + } + } +} + +/// +/// Drains onto the durable , +/// owns the one-time schema bootstrap, and sweeps audit rows past their retention window. +/// +/// +/// Batching is the point: up to buffered events are committed in a +/// single transaction, so a burst of constraint denials costs a handful of commits instead of +/// one per denied tag. The bootstrap runs here — before the writer starts enqueueing — so no +/// audit write ever pays a CREATE TABLE IF NOT EXISTS round-trip. +/// +/// The channel writer whose buffered events are drained. +/// The durable sink events are committed to. +/// Security options carrying the audit retention window. +/// Clock used for the retention cutoff and sweep interval. +/// Logger for bootstrap, drain and sweep diagnostics. +public sealed class AuditDrainService( + ChannelAuditWriter writer, + IAuditEventSink sink, + SecurityOptions security, + TimeProvider timeProvider, + ILogger logger) : BackgroundService +{ + /// Maximum number of audit events committed in one transaction per drain pass. + public const int MaxBatchSize = 64; + + /// How often the retention sweep runs while the gateway is up. + public static readonly TimeSpan RetentionSweepInterval = TimeSpan.FromHours(1); + + /// Upper bound on how long shutdown waits for the remaining buffered events. + private static readonly TimeSpan ShutdownDrainCap = TimeSpan.FromSeconds(2); + + /// + /// Bootstraps the audit table, runs one retention sweep, then attaches the drain so the + /// writer switches from synchronous write-through to enqueueing. + /// + /// Token to observe for cancellation. + /// A task that represents the asynchronous operation. + public override async Task StartAsync(CancellationToken cancellationToken) + { + try + { + await sink.EnsureInitializedAsync(cancellationToken).ConfigureAwait(false); + await SweepRetentionAsync(cancellationToken).ConfigureAwait(false); + } + catch (Exception exception) + { + // Audit is best-effort: a bootstrap failure must not take the gateway down. The + // sink's own latch will retry the schema check on the first write. + logger.LogWarning(exception, "Audit store bootstrap failed; audit writes will retry the schema check."); + } + + writer.AttachDrain(); + + await base.StartAsync(cancellationToken).ConfigureAwait(false); + } + + /// + /// Detaches the drain (so late writes go straight to the sink) and gives the buffered + /// events a bounded window to reach the store. + /// + /// Token to observe for cancellation. + /// A task that represents the asynchronous operation. + public override async Task StopAsync(CancellationToken cancellationToken) + { + writer.DetachDrain(); + writer.CompleteWriting(); + + await base.StopAsync(cancellationToken).ConfigureAwait(false); + + using CancellationTokenSource drainCap = new(ShutdownDrainCap); + try + { + await DrainPendingAsync(drainCap.Token).ConfigureAwait(false); + } + catch (OperationCanceledException) + { + logger.LogWarning( + "Shutdown drain exceeded {CapSeconds}s; remaining buffered audit events were not persisted.", + ShutdownDrainCap.TotalSeconds); + } + } + + /// + /// Commits every event currently buffered, in batches of at most . + /// + /// Token to observe for cancellation. + /// The number of events persisted. + public async Task DrainPendingAsync(CancellationToken cancellationToken) + { + int persisted = 0; + List batch = new(MaxBatchSize); + + while (!cancellationToken.IsCancellationRequested) + { + batch.Clear(); + while (batch.Count < MaxBatchSize && writer.Reader.TryRead(out AuditEvent? auditEvent)) + { + batch.Add(auditEvent); + } + + if (batch.Count == 0) + { + break; + } + + try + { + await sink.InsertBatchAsync(batch, cancellationToken).ConfigureAwait(false); + persisted += batch.Count; + } + catch (Exception exception) + { + // Best-effort: the batch is lost, but the drain keeps going so one poisoned + // batch cannot stall every later audit event behind it. + logger.LogWarning( + exception, + "Failed to persist a batch of {Count} audit events; audit writes are best-effort and were suppressed.", + batch.Count); + } + } + + return persisted; + } + + /// + /// Deletes audit rows older than MxGateway:Security:AuditRetentionDays, and reports + /// any audit events dropped by channel pressure since the last sweep. + /// + /// Token to observe for cancellation. + /// A task that represents the asynchronous operation. + public async Task SweepRetentionAsync(CancellationToken cancellationToken) + { + DateTimeOffset cutoff = timeProvider.GetUtcNow() - TimeSpan.FromDays(security.AuditRetentionDays); + + try + { + int deleted = await sink.DeleteOlderThanAsync(cutoff, cancellationToken).ConfigureAwait(false); + if (deleted > 0) + { + logger.LogInformation( + "Audit retention sweep removed {Deleted} events older than {Cutoff:o} ({RetentionDays} days).", + deleted, + cutoff, + security.AuditRetentionDays); + } + } + catch (Exception exception) + { + logger.LogWarning(exception, "Audit retention sweep failed; it will be retried on the next interval."); + } + + long dropped = writer.DroppedCount; + if (dropped > 0) + { + logger.LogWarning( + "{Dropped} audit events have been dropped since startup because the audit channel was full.", + dropped); + } + } + + /// + protected override async Task ExecuteAsync(CancellationToken stoppingToken) + { + await Task.WhenAll( + DrainLoopAsync(stoppingToken), + RetentionLoopAsync(stoppingToken)).ConfigureAwait(false); + } + + private async Task DrainLoopAsync(CancellationToken stoppingToken) + { + try + { + while (await writer.Reader.WaitToReadAsync(stoppingToken).ConfigureAwait(false)) + { + await DrainPendingAsync(stoppingToken).ConfigureAwait(false); + } + } + catch (OperationCanceledException) + { + // Shutdown; StopAsync performs the final bounded drain. + } + } + + private async Task RetentionLoopAsync(CancellationToken stoppingToken) + { + using PeriodicTimer timer = new(RetentionSweepInterval, timeProvider); + + try + { + while (await timer.WaitForNextTickAsync(stoppingToken).ConfigureAwait(false)) + { + await SweepRetentionAsync(stoppingToken).ConfigureAwait(false); + } + } + catch (OperationCanceledException) + { + // Shutdown. + } + } +} diff --git a/src/ZB.MOM.WW.MxGateway.Server/Security/Audit/SqliteCanonicalAuditStore.cs b/src/ZB.MOM.WW.MxGateway.Server/Security/Audit/SqliteCanonicalAuditStore.cs index a5ff0b8..db83c62 100644 --- a/src/ZB.MOM.WW.MxGateway.Server/Security/Audit/SqliteCanonicalAuditStore.cs +++ b/src/ZB.MOM.WW.MxGateway.Server/Security/Audit/SqliteCanonicalAuditStore.cs @@ -5,6 +5,41 @@ using ZB.MOM.WW.Auth.ApiKeys.Sqlite; namespace ZB.MOM.WW.MxGateway.Server.Security.Audit; +/// +/// Durable sink the audit pipeline persists canonical s through. +/// It exists so the write path () and the batching drain +/// () depend on the storage contract rather than on the +/// concrete . +/// +public interface IAuditEventSink +{ + /// + /// Bootstraps the backing storage. Called once at startup so no write path pays a schema + /// round-trip; implementations must be idempotent and safe to call concurrently. + /// + /// Token to observe for cancellation. + /// A task that represents the asynchronous operation. + Task EnsureInitializedAsync(CancellationToken cancellationToken); + + /// Persists a single canonical audit event. + /// The canonical event to persist. + /// Token to observe for cancellation. + /// A task that represents the asynchronous operation. + Task InsertAsync(AuditEvent auditEvent, CancellationToken cancellationToken); + + /// Persists a batch of canonical audit events as one unit of work. + /// The canonical events to persist. + /// Token to observe for cancellation. + /// A task that represents the asynchronous operation. + Task InsertBatchAsync(IReadOnlyList auditEvents, CancellationToken cancellationToken); + + /// Deletes every audit row that occurred strictly before . + /// The retention cutoff; rows older than this are removed. + /// Token to observe for cancellation. + /// The number of rows deleted. + Task DeleteOlderThanAsync(DateTimeOffset cutoffUtc, CancellationToken cancellationToken); +} + /// /// MxGateway-owned, append-only SQLite store for canonical /// s. It writes to a NEW audit_event table in the @@ -18,11 +53,20 @@ namespace ZB.MOM.WW.MxGateway.Server.Security.Audit; /// IApiKeyAuditStore registration is overridden by /// , which forwards onto this store via /// . The library's schema_version / -/// api_key_audit tables are not touched here; the audit_event table is -/// created idempotently (CREATE TABLE IF NOT EXISTS) on each write so it -/// self-bootstraps regardless of migration ordering. +/// api_key_audit tables are not touched here. +/// +/// The audit_event table is created idempotently, but the CREATE TABLE IF NOT +/// EXISTS is latched: runs +/// once at startup, and every later insert/list/delete +/// then skips the DDL round-trip. Keeping the (now free) check on each path rather than +/// dropping it means the store still self-bootstraps for callers that use it without the +/// hosted drain — the apikey CLI and the DI-only tests — regardless of migration +/// ordering. The latch is deliberately racy: a lost race merely re-runs an idempotent +/// CREATE TABLE IF NOT EXISTS, and a failure leaves the latch open so the next call +/// retries. +/// /// -public sealed class SqliteCanonicalAuditStore(AuthSqliteConnectionFactory connectionFactory) +public sealed class SqliteCanonicalAuditStore(AuthSqliteConnectionFactory connectionFactory) : IAuditEventSink { private const string CreateTableSql = """ @@ -40,14 +84,107 @@ public sealed class SqliteCanonicalAuditStore(AuthSqliteConnectionFactory connec ); """; - /// Inserts a canonical audit event into the audit_event table. - /// The canonical event to persist. - /// Token to observe for cancellation. - /// A task that represents the asynchronous operation. - public async Task InsertAsync(AuditEvent auditEvent, CancellationToken cancellationToken) + private const string InsertSql = + """ + INSERT INTO audit_event + (event_id, occurred_at_utc, actor, action, outcome, + category, target, source_node, correlation_id, details_json) + VALUES + ($event_id, $occurred_at_utc, $actor, $action, $outcome, + $category, $target, $source_node, $correlation_id, $details_json); + """; + + /// 0 until the audit_event table has been created at least once by this instance. + private int _tableEnsured; + + /// + public async Task EnsureInitializedAsync(CancellationToken cancellationToken) + { + await using SqliteConnection connection = + await connectionFactory.OpenConnectionAsync(cancellationToken).ConfigureAwait(false); + + await EnsureTableAsync(connection, cancellationToken).ConfigureAwait(false); + } + + /// + public Task InsertAsync(AuditEvent auditEvent, CancellationToken cancellationToken) { ArgumentNullException.ThrowIfNull(auditEvent); + return InsertBatchAsync([auditEvent], cancellationToken); + } + + /// + /// + /// One connection, one transaction and one prepared command for the whole batch: the drain + /// pays a single commit for up to events rather + /// than one round-trip per event. + /// + public async Task InsertBatchAsync(IReadOnlyList auditEvents, CancellationToken cancellationToken) + { + ArgumentNullException.ThrowIfNull(auditEvents); + + if (auditEvents.Count == 0) + { + return; + } + + await using SqliteConnection connection = + await connectionFactory.OpenConnectionAsync(cancellationToken).ConfigureAwait(false); + + await EnsureTableAsync(connection, cancellationToken).ConfigureAwait(false); + + await using SqliteTransaction transaction = + (SqliteTransaction)await connection.BeginTransactionAsync(cancellationToken).ConfigureAwait(false); + + await using (SqliteCommand command = connection.CreateCommand()) + { + command.Transaction = transaction; + command.CommandText = InsertSql; + + SqliteParameter eventId = command.Parameters.Add("$event_id", SqliteType.Text); + SqliteParameter occurredAtUtc = command.Parameters.Add("$occurred_at_utc", SqliteType.Text); + SqliteParameter actor = command.Parameters.Add("$actor", SqliteType.Text); + SqliteParameter action = command.Parameters.Add("$action", SqliteType.Text); + SqliteParameter outcome = command.Parameters.Add("$outcome", SqliteType.Text); + SqliteParameter category = command.Parameters.Add("$category", SqliteType.Text); + SqliteParameter target = command.Parameters.Add("$target", SqliteType.Text); + SqliteParameter sourceNode = command.Parameters.Add("$source_node", SqliteType.Text); + SqliteParameter correlationId = command.Parameters.Add("$correlation_id", SqliteType.Text); + SqliteParameter detailsJson = command.Parameters.Add("$details_json", SqliteType.Text); + + foreach (AuditEvent auditEvent in auditEvents) + { + ArgumentNullException.ThrowIfNull(auditEvent); + + eventId.Value = auditEvent.EventId.ToString(); + occurredAtUtc.Value = auditEvent.OccurredAtUtc.ToString("O", CultureInfo.InvariantCulture); + actor.Value = auditEvent.Actor; + action.Value = auditEvent.Action; + outcome.Value = auditEvent.Outcome.ToString(); + category.Value = (object?)auditEvent.Category ?? DBNull.Value; + target.Value = (object?)auditEvent.Target ?? DBNull.Value; + sourceNode.Value = (object?)auditEvent.SourceNode ?? DBNull.Value; + correlationId.Value = (object?)auditEvent.CorrelationId?.ToString() ?? DBNull.Value; + detailsJson.Value = (object?)auditEvent.DetailsJson ?? DBNull.Value; + + await command.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false); + } + } + + await transaction.CommitAsync(cancellationToken).ConfigureAwait(false); + } + + /// + /// + /// The comparison goes through SQLite's datetime() rather than comparing the stored + /// ISO-8601 text directly: rows written from a non-UTC (the + /// library's CreatedUtc is caller-supplied) do not sort lexicographically against a + /// UTC cutoff. A row whose timestamp datetime() cannot parse yields NULL and is + /// therefore never deleted — audit that cannot be dated is kept, not swept. + /// + public async Task DeleteOlderThanAsync(DateTimeOffset cutoffUtc, CancellationToken cancellationToken) + { await using SqliteConnection connection = await connectionFactory.OpenConnectionAsync(cancellationToken).ConfigureAwait(false); @@ -56,25 +193,14 @@ public sealed class SqliteCanonicalAuditStore(AuthSqliteConnectionFactory connec await using SqliteCommand command = connection.CreateCommand(); command.CommandText = """ - INSERT INTO audit_event - (event_id, occurred_at_utc, actor, action, outcome, - category, target, source_node, correlation_id, details_json) - VALUES - ($event_id, $occurred_at_utc, $actor, $action, $outcome, - $category, $target, $source_node, $correlation_id, $details_json); + DELETE FROM audit_event + WHERE datetime(occurred_at_utc) < datetime($cutoff); """; - command.Parameters.AddWithValue("$event_id", auditEvent.EventId.ToString()); - command.Parameters.AddWithValue("$occurred_at_utc", auditEvent.OccurredAtUtc.ToString("O", CultureInfo.InvariantCulture)); - command.Parameters.AddWithValue("$actor", auditEvent.Actor); - command.Parameters.AddWithValue("$action", auditEvent.Action); - command.Parameters.AddWithValue("$outcome", auditEvent.Outcome.ToString()); - command.Parameters.AddWithValue("$category", (object?)auditEvent.Category ?? DBNull.Value); - command.Parameters.AddWithValue("$target", (object?)auditEvent.Target ?? DBNull.Value); - command.Parameters.AddWithValue("$source_node", (object?)auditEvent.SourceNode ?? DBNull.Value); - command.Parameters.AddWithValue("$correlation_id", (object?)auditEvent.CorrelationId?.ToString() ?? DBNull.Value); - command.Parameters.AddWithValue("$details_json", (object?)auditEvent.DetailsJson ?? DBNull.Value); + command.Parameters.AddWithValue( + "$cutoff", + cutoffUtc.ToUniversalTime().ToString("O", CultureInfo.InvariantCulture)); - await command.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false); + return await command.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false); } /// Returns the most recent canonical audit events, newest first. @@ -128,11 +254,20 @@ public sealed class SqliteCanonicalAuditStore(AuthSqliteConnectionFactory connec return events; } - private static async Task EnsureTableAsync(SqliteConnection connection, CancellationToken cancellationToken) + // Latched bootstrap: after the first success this is a single volatile read, so the DDL + // round-trip is paid once per process rather than once per audit write. + private async Task EnsureTableAsync(SqliteConnection connection, CancellationToken cancellationToken) { + if (Volatile.Read(ref _tableEnsured) == 1) + { + return; + } + await using SqliteCommand command = connection.CreateCommand(); command.CommandText = CreateTableSql; await command.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false); + + Volatile.Write(ref _tableEnsured, 1); } private static DateTimeOffset ParseUtc(string value) => diff --git a/src/ZB.MOM.WW.MxGateway.Server/Security/Authentication/AuthStoreServiceCollectionExtensions.cs b/src/ZB.MOM.WW.MxGateway.Server/Security/Authentication/AuthStoreServiceCollectionExtensions.cs index 5c3d40a..6722fa6 100644 --- a/src/ZB.MOM.WW.MxGateway.Server/Security/Authentication/AuthStoreServiceCollectionExtensions.cs +++ b/src/ZB.MOM.WW.MxGateway.Server/Security/Authentication/AuthStoreServiceCollectionExtensions.cs @@ -99,16 +99,40 @@ public static class AuthStoreServiceCollectionExtensions services.AddSingleton(sp => new SqliteCanonicalAuditStore(sp.GetRequiredService())); + services.AddSingleton(sp => sp.GetRequiredService()); // Resolve the logger defensively: the production host always registers ILogger, but the // DI-only auth/CLI/dashboard unit tests build a bare ServiceCollection without AddLogging(). // Fall back to NullLogger there so the audit writer (and the IApiKeyAuditStore override that // depends on it) still resolve. The write path is best-effort regardless. - services.AddSingleton(sp => + services.AddSingleton(sp => new CanonicalAuditWriter( - sp.GetRequiredService(), + sp.GetRequiredService(), sp.GetService>() ?? Microsoft.Extensions.Logging.Abstractions.NullLogger.Instance)); + // The registered IAuditWriter is the bounded, asynchronous one: audit producers — above + // all IConstraintEnforcer.RecordDenialAsync, which fires once per denied tag inside bulk + // RPC loops — enqueue and return instead of awaiting a SQLite insert each. No producer + // signature changes; the seam is entirely here. AuditDrainService batches the buffered + // events onto the sink, owns the one-time schema bootstrap and sweeps expired rows. Where + // no hosted service runs (the `apikey` CLI, the DI-only tests) the channel writer falls + // back to CanonicalAuditWriter's synchronous path, so audit is never silently buffered + // into a channel nobody drains. + services.AddSingleton(sp => + new ChannelAuditWriter( + sp.GetRequiredService(), + sp.GetService>() + ?? Microsoft.Extensions.Logging.Abstractions.NullLogger.Instance)); + services.AddSingleton(sp => sp.GetRequiredService()); + services.AddSingleton(sp => new AuditDrainService( + sp.GetRequiredService(), + sp.GetRequiredService(), + security, + sp.GetService() ?? TimeProvider.System, + sp.GetService>() + ?? Microsoft.Extensions.Logging.Abstractions.NullLogger.Instance)); + services.AddHostedService(sp => sp.GetRequiredService()); + // OVERRIDE the library's IApiKeyAuditStore (AddZbApiKeyAuth registered the library's // SqliteApiKeyAuditStore via TryAddSingleton) with an adapter that canonicalizes every // library-emitted ApiKeyAuditEntry onto AuditEvent and forwards it through IAuditWriter. diff --git a/src/ZB.MOM.WW.MxGateway.Tests/Security/Audit/ChannelAuditWriterTests.cs b/src/ZB.MOM.WW.MxGateway.Tests/Security/Audit/ChannelAuditWriterTests.cs new file mode 100644 index 0000000..c96dce5 --- /dev/null +++ b/src/ZB.MOM.WW.MxGateway.Tests/Security/Audit/ChannelAuditWriterTests.cs @@ -0,0 +1,271 @@ +using Microsoft.Extensions.Logging.Abstractions; +using Microsoft.Extensions.Time.Testing; +using ZB.MOM.WW.Audit; +using ZB.MOM.WW.Auth.ApiKeys.Sqlite; +using ZB.MOM.WW.MxGateway.Server.Configuration; +using ZB.MOM.WW.MxGateway.Server.Security.Audit; +using ZB.MOM.WW.MxGateway.Tests.Security.Authentication; + +namespace ZB.MOM.WW.MxGateway.Tests.Security.Audit; + +/// +/// Tests the bounded, asynchronous audit path: (enqueue-only +/// once a drain is attached, drops rather than blocks when the channel is full) and +/// (batched drain, one-time table bootstrap, retention sweep). +/// The channel makes the already-documented best-effort audit contract explicit and bounded: +/// a partially denied bulk RPC no longer pays a SQLite round-trip per denied tag. +/// +public sealed class ChannelAuditWriterTests : IDisposable +{ + private readonly List _tempDirectories = []; + + /// + /// With a drain attached, only enqueues — the + /// sink is not touched until the drain runs, and the event arrives once it does. + /// + /// A task that represents the asynchronous operation. + [Fact] + public async Task WriteAsync_WithDrainAttached_EnqueuesOnlyAndReachesSinkAfterDrain() + { + CountingAuditSink sink = new(); + (ChannelAuditWriter writer, AuditDrainService drain) = CreateWriterAndDrain(sink); + writer.AttachDrain(); + + await writer.WriteAsync(MakeEvent("constraint-denied"), CancellationToken.None); + + // Enqueue-only: nothing has reached the sink yet. + Assert.Equal(0, sink.InsertBatchCalls); + Assert.Empty(sink.Events); + + int drained = await drain.DrainPendingAsync(CancellationToken.None); + + Assert.Equal(1, drained); + Assert.Equal(1, sink.InsertBatchCalls); + Assert.Equal("constraint-denied", Assert.Single(sink.Events).Action); + } + + /// + /// A drain pass batches up to events into one + /// sink call, so a partially denied bulk RPC costs a handful of transactions, not one per tag. + /// + /// A task that represents the asynchronous operation. + [Fact] + public async Task DrainPendingAsync_BatchesEventsIntoBoundedInsertCalls() + { + CountingAuditSink sink = new(); + (ChannelAuditWriter writer, AuditDrainService drain) = CreateWriterAndDrain(sink); + writer.AttachDrain(); + + const int eventCount = 150; + for (int index = 0; index < eventCount; index++) + { + await writer.WriteAsync(MakeEvent($"denied-{index}"), CancellationToken.None); + } + + int drained = await drain.DrainPendingAsync(CancellationToken.None); + + Assert.Equal(eventCount, drained); + Assert.Equal(eventCount, sink.Events.Count); + // 150 events at a batch size of 64 → 3 transactions (64 + 64 + 22), not 150. + Assert.Equal(3, sink.InsertBatchCalls); + Assert.All(sink.BatchSizes, size => Assert.True(size <= AuditDrainService.MaxBatchSize)); + } + + /// + /// When the bounded channel is full the write is dropped rather than blocking the caller, + /// and the drop is counted. A stalled or slow audit database must never stall an RPC. + /// + /// A task that represents the asynchronous operation. + [Fact] + public async Task WriteAsync_WhenChannelFull_DropsWriteAndCountsItWithoutBlocking() + { + CountingAuditSink sink = new(); + (ChannelAuditWriter writer, _) = CreateWriterAndDrain(sink); + writer.AttachDrain(); + + const int overflow = 32; + for (int index = 0; index < ChannelAuditWriter.ChannelCapacity + overflow; index++) + { + // Every call must complete synchronously: the channel never blocks a producer. + Task write = writer.WriteAsync(MakeEvent($"denied-{index}"), CancellationToken.None); + Assert.True(write.IsCompletedSuccessfully); + } + + Assert.Equal(overflow, writer.DroppedCount); + // The channel still holds exactly its capacity; nothing reached the sink (no drain ran). + Assert.Equal(0, sink.InsertBatchCalls); + + await Task.CompletedTask; + } + + /// + /// With no drain attached (the apikey CLI, DI-only tests, post-shutdown) the writer + /// writes through to the sink instead of enqueueing into a channel nobody will ever read. + /// + /// A task that represents the asynchronous operation. + [Fact] + public async Task WriteAsync_WithNoDrainAttached_WritesThroughToSink() + { + CountingAuditSink sink = new(); + (ChannelAuditWriter writer, _) = CreateWriterAndDrain(sink); + + await writer.WriteAsync(MakeEvent("dashboard-create-key"), CancellationToken.None); + + Assert.Equal("dashboard-create-key", Assert.Single(sink.Events).Action); + } + + /// + /// The retention sweep deletes rows older than AuditRetentionDays, measured from the + /// injected clock, and runs once at startup after the one-time table bootstrap. + /// + /// A task that represents the asynchronous operation. + [Fact] + public async Task StartAsync_BootstrapsTableThenSweepsRetentionAtConfiguredCutoff() + { + CountingAuditSink sink = new(); + FakeTimeProvider clock = new(new DateTimeOffset(2026, 8, 15, 12, 0, 0, TimeSpan.Zero)); + (_, AuditDrainService drain) = CreateWriterAndDrain( + sink, + new SecurityOptions { AuditRetentionDays = 30 }, + clock); + + await drain.StartAsync(CancellationToken.None); + await drain.StopAsync(CancellationToken.None); + + Assert.Equal(1, sink.EnsureInitializedCalls); + Assert.Equal(new DateTimeOffset(2026, 7, 16, 12, 0, 0, TimeSpan.Zero), Assert.Single(sink.DeleteCutoffs)); + } + + /// + /// The real SQLite sink deletes only rows older than the cutoff, and its bootstrap latch + /// means the CREATE TABLE round-trip is not repeated on every subsequent write. + /// + /// A task that represents the asynchronous operation. + [Fact] + public async Task SqliteStore_DeleteOlderThan_RemovesOnlyRowsBeyondTheRetentionWindow() + { + SqliteCanonicalAuditStore store = CreateStore(); + DateTimeOffset now = new(2026, 8, 15, 12, 0, 0, TimeSpan.Zero); + + await store.InsertBatchAsync( + [ + MakeEvent("stale", now.AddDays(-120)), + // A non-UTC offset must still compare correctly against the cutoff. + MakeEvent("stale-offset", now.AddDays(-100).ToOffset(TimeSpan.FromHours(-5))), + MakeEvent("fresh", now.AddDays(-1)), + ], + CancellationToken.None); + + int deleted = await store.DeleteOlderThanAsync(now.AddDays(-90), CancellationToken.None); + + Assert.Equal(2, deleted); + Assert.Equal("fresh", Assert.Single(await store.ListRecentAsync(10, CancellationToken.None)).Action); + } + + private static (ChannelAuditWriter Writer, AuditDrainService Drain) CreateWriterAndDrain( + IAuditEventSink sink, + SecurityOptions? security = null, + TimeProvider? timeProvider = null) + { + ChannelAuditWriter writer = new( + new CanonicalAuditWriter(sink, NullLogger.Instance), + NullLogger.Instance); + AuditDrainService drain = new( + writer, + sink, + security ?? new SecurityOptions(), + timeProvider ?? TimeProvider.System, + NullLogger.Instance); + return (writer, drain); + } + + private SqliteCanonicalAuditStore CreateStore() + { + TempDatabaseDirectory directory = TempDatabaseDirectory.Create("mxgateway-channel-audit"); + _tempDirectories.Add(directory); + return new SqliteCanonicalAuditStore(new AuthSqliteConnectionFactory(directory.DatabasePath())); + } + + private static AuditEvent MakeEvent(string action, DateTimeOffset? occurredAtUtc = null) => new() + { + EventId = Guid.NewGuid(), + OccurredAtUtc = occurredAtUtc ?? DateTimeOffset.UtcNow, + Actor = "operator01", + Action = action, + Outcome = AuditOutcome.Denied, + Category = "ApiKey", + }; + + /// Clears SQLite pools and deletes every temporary directory created by this test. + public void Dispose() + { + foreach (TempDatabaseDirectory directory in _tempDirectories) + { + directory.Dispose(); + } + + _tempDirectories.Clear(); + } + + /// + /// In-memory that records every call, so a test can prove the + /// write path did NOT touch the sink and that the drain batched what it did write. + /// + private sealed class CountingAuditSink : IAuditEventSink + { + private readonly Lock _gate = new(); + + /// Gets the events handed to the sink, in the order they were written. + public List Events { get; } = []; + + /// Gets the size of each batch the sink was asked to insert. + public List BatchSizes { get; } = []; + + /// Gets the cutoffs the sink was asked to delete below. + public List DeleteCutoffs { get; } = []; + + /// Gets the number of calls. + public int InsertBatchCalls => BatchSizes.Count; + + /// Gets the number of calls. + public int EnsureInitializedCalls { get; private set; } + + /// + public Task EnsureInitializedAsync(CancellationToken cancellationToken) + { + lock (_gate) + { + EnsureInitializedCalls++; + } + + return Task.CompletedTask; + } + + /// + public Task InsertAsync(AuditEvent auditEvent, CancellationToken cancellationToken) => + InsertBatchAsync([auditEvent], cancellationToken); + + /// + public Task InsertBatchAsync(IReadOnlyList auditEvents, CancellationToken cancellationToken) + { + lock (_gate) + { + BatchSizes.Add(auditEvents.Count); + Events.AddRange(auditEvents); + } + + return Task.CompletedTask; + } + + /// + public Task DeleteOlderThanAsync(DateTimeOffset cutoffUtc, CancellationToken cancellationToken) + { + lock (_gate) + { + DeleteCutoffs.Add(cutoffUtc); + } + + return Task.FromResult(0); + } + } +}