feat(audit): off-box chain-head export, verify-heads, maintenance login (ScadaBridge#107)

Item 1 (owner decision: no signing). AuditLog:HashChain:HeadExport (default off): both centrals periodically read every month's chain head (LastSeq, LastHash, purge anchor, chained rows, tombstones; one snapshot under the chain lock SHARED) and append the changed heads to a sink the DB login cannot write - File (append-only JSONL per node per day, local or UNC) or Syslog (RFC 5424 over UDP / TCP octet counting). `audit verify-heads --path` reads those records back (file or syslog-collector lines) and the new VerifyAuditChainHeads command checks each against the database: Matches / Extended / Purged, or Mismatch / Regressed / MissingHead / MissingLink / CountMismatch (exit 3). AuditChainHeadStoreTests forges a chain with the production hasher: verify-chain says Intact, verify-heads says Mismatch. Item 3. AuditLog:MaintenanceConnectionString (${secret:}-capable, unset = today's single login) runs the chain job, chain verify / head export, the partition switch-out + archive and the per-channel purge on AuditMaintenanceContext (keyed audit-maintenance repository/archive, purge actor prefers them). AuditMaintenanceLoginTests runs the hardened setup on SQL Server: writer (datareader/datawriter/audit_writer) ingests and is refused chaining and deleting; the chainer+purger login does the full cycle. Two-login grants documented in Component-AuditLog.md. Claude-Session: https://claude.ai/code/session_01GT67AEHd2Acv1YPwGfXEoV
2026-11-09 23:16:05 -05:00
parent 5e7d5389d4
commit c4d7ec9b96
40 changed files with 2769 additions and 43 deletions
@@ -792,4 +792,34 @@ public class AuditLogPurgeActorTests : TestKit, IClassFixture<MsSqlMigrationFixt

        StopPurgeActor(actor);
    }
+
+    /// <summary>
+    /// ScadaBridge#107 item 3: when the host registered the maintenance-connection repository
+    /// (keyed <c>audit-maintenance</c>), the purge runs on it, never on the runtime one.
+    /// </summary>
+    [Fact]
+    public void Tick_UsesTheMaintenanceConnectionRepository_WhenRegistered()
+    {
+        var runtime = new RecordingRepo { Boundaries = [new DateTime(2025, 1, 1, 0, 0, 0, DateTimeKind.Utc)] };
+        var maintenance = new RecordingRepo { Boundaries = [new DateTime(2025, 1, 1, 0, 0, 0, DateTimeKind.Utc)] };
+        var services = new ServiceCollection();
+        services.AddScoped<IAuditLogRepository>(_ => runtime);
+        services.AddKeyedScoped<IAuditLogRepository>(
+            ZB.MOM.WW.ScadaBridge.ConfigurationDatabase.Maintenance.AuditMaintenanceContext.ServiceKey, (_, _) => maintenance);
+        var sp = services.BuildServiceProvider();
+        var actor = Sys.ActorOf(Props.Create(() => new AuditLogPurgeActor(
+            sp,
+            Options.Create(FastTickOptions()),
+            Options.Create(new AuditLogOptions()),
+            NullLogger<AuditLogPurgeActor>.Instance)));
+
+        AwaitAssert(
+            () => Assert.NotEmpty(maintenance.SwitchedBoundaries),
+            duration: TimeSpan.FromSeconds(3),
+            interval: TimeSpan.FromMilliseconds(50));
+        Assert.Empty(runtime.ThresholdQueries);
+        Assert.Empty(runtime.SwitchedBoundaries);
+
+        StopPurgeActor(actor);
+    }
 }
@@ -0,0 +1,164 @@
+using Microsoft.Extensions.DependencyInjection;
+using Microsoft.Extensions.Logging.Abstractions;
+using Microsoft.Extensions.Options;
+using ZB.MOM.WW.ScadaBridge.AuditLog.Central;
+using ZB.MOM.WW.ScadaBridge.AuditLog.Central.HeadExport;
+using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Repositories;
+using ZB.MOM.WW.ScadaBridge.Commons.Types.Audit;
+
+namespace ZB.MOM.WW.ScadaBridge.AuditLog.Tests.Central.HeadExport;
+
+/// <summary>
+/// <see cref="AuditChainHeadExportService"/> (ScadaBridge#107 item 1): first tick exports every
+/// head, later ticks only the changed ones, a failed write is retried with the same heads, and
+/// the options validator guards each sink's settings.
+/// </summary>
+public sealed class AuditChainHeadExportServiceTests
+{
+    private sealed class FakeStore : IAuditChainHeadStore
+    {
+        public List<AuditChainHeadSnapshot> Heads { get; } = [];
+
+        public Task<IReadOnlyList<AuditChainHeadSnapshot>> ReadHeadsAsync(CancellationToken ct = default)
+            => Task.FromResult<IReadOnlyList<AuditChainHeadSnapshot>>(Heads.ToList());
+
+        public Task<AuditChainHeadVerificationResult> VerifyExportedHeadsAsync(
+            IReadOnlyList<AuditChainHeadRecord> records, CancellationToken ct = default) => throw new NotSupportedException();
+    }
+
+    private sealed class RecordingSink : IAuditChainHeadSink
+    {
+        public List<IReadOnlyList<AuditChainHeadRecord>> Writes { get; } = [];
+
+        public bool Fail { get; set; }
+
+        public string Describe => "recording";
+
+        public Task WriteAsync(IReadOnlyList<AuditChainHeadRecord> records, CancellationToken ct)
+        {
+            if (Fail)
+            {
+                throw new IOException("share offline");
+            }
+
+            Writes.Add(records);
+            return Task.CompletedTask;
+        }
+    }
+
+    private static AuditChainHeadSnapshot Head(int month, long seq, long rows) => new(
+        new DateTime(2026, month, 1, 0, 0, 0, DateTimeKind.Utc), seq, Enumerable.Repeat((byte)seq, 32).ToArray(),
+        null, null, rows, 0, new DateTime(2026, month, 2, 0, 0, 0, DateTimeKind.Utc));
+
+    private static (AuditChainHeadExportService Service, FakeStore Store, RecordingSink Sink) Create(bool exportUnchanged = false)
+    {
+        var store = new FakeStore();
+        var sink = new RecordingSink();
+        var services = new ServiceCollection();
+        services.AddScoped<IAuditChainHeadStore>(_ => store);
+        var options = Options.Create(new AuditLogHashChainOptions
+        {
+            HeadExport = new AuditChainHeadExportOptions { Enabled = true, ExportUnchanged = exportUnchanged },
+        });
+        var service = new AuditChainHeadExportService(
+            services.BuildServiceProvider().GetRequiredService<IServiceScopeFactory>(),
+            options, sink, "central-a", NullLogger<AuditChainHeadExportService>.Instance);
+        return (service, store, sink);
+    }
+
+    /// <summary>First tick: every head; next tick: only the head that moved; unchanged: nothing.</summary>
+    /// <returns>A task that represents the asynchronous operation.</returns>
+    [Fact]
+    public async Task Tick_ExportsAllHeadsFirst_ThenOnlyChangedOnes()
+    {
+        var (service, store, sink) = Create();
+        store.Heads.AddRange([Head(4, 10, 10), Head(5, 20, 20)]);
+
+        Assert.Equal(2, await service.TickAsync(default));
+        Assert.Equal(0, await service.TickAsync(default));
+
+        store.Heads[1] = Head(5, 25, 25);
+        Assert.Equal(1, await service.TickAsync(default));
+
+        // A per-channel purge (row -> tombstone) keeps rows + tombstones, so it is not a change;
+        // a row disappearing without a tombstone is.
+        store.Heads[0] = Head(4, 10, 9);
+        Assert.Equal(1, await service.TickAsync(default));
+
+        Assert.Equal(3, sink.Writes.Count);
+        Assert.Equal(["2026-04", "2026-05"], sink.Writes[0].Select(r => r.Month));
+        var moved = Assert.Single(sink.Writes[1]);
+        Assert.Equal(("2026-05", 25L, string.Concat(Enumerable.Repeat("19", 32))), (moved.Month, moved.LastSeq, moved.LastHash));
+        Assert.Equal("central-a", moved.Node);
+        Assert.Equal(AuditChainHeadRecord.CurrentVersion, moved.V);
+    }
+
+    /// <summary>ExportUnchanged re-exports every head every tick.</summary>
+    /// <returns>A task that represents the asynchronous operation.</returns>
+    [Fact]
+    public async Task Tick_WithExportUnchanged_ExportsEveryHeadEveryTick()
+    {
+        var (service, store, _) = Create(exportUnchanged: true);
+        store.Heads.AddRange([Head(4, 10, 10), Head(5, 20, 20)]);
+
+        Assert.Equal(2, await service.TickAsync(default));
+        Assert.Equal(2, await service.TickAsync(default));
+    }
+
+    /// <summary>A failed write is contained and the same heads go out on the next tick.</summary>
+    /// <returns>A task that represents the asynchronous operation.</returns>
+    [Fact]
+    public async Task Tick_AfterAFailedWrite_RetriesTheSameHeads()
+    {
+        var (service, store, sink) = Create();
+        store.Heads.Add(Head(4, 10, 10));
+        sink.Fail = true;
+
+        Assert.Equal(0, await service.TickAsync(default));
+        sink.Fail = false;
+        Assert.Equal(1, await service.TickAsync(default));
+        Assert.Equal(10, Assert.Single(Assert.Single(sink.Writes)).LastSeq);
+    }
+
+    /// <summary>The validator ignores a disabled export and checks each sink's settings when enabled.</summary>
+    /// <param name="configure">Which settings to apply.</param>
+    /// <param name="expectedError">A fragment of the expected failure, or null when valid.</param>
+    [Theory]
+    [InlineData("disabled-bad", null)]
+    [InlineData("file-ok", null)]
+    [InlineData("file-relative", "File:Directory")]
+    [InlineData("file-missing", "File:Directory")]
+    [InlineData("syslog-ok", null)]
+    [InlineData("syslog-no-host", "Syslog:Host")]
+    [InlineData("syslog-bad-port", "Syslog:Port")]
+    [InlineData("syslog-bad-facility", "Syslog:Facility")]
+    [InlineData("interval-zero", "IntervalMinutes")]
+    public void Validator_ChecksTheSelectedSink(string configure, string? expectedError)
+    {
+        var root = OperatingSystem.IsWindows() ? @"C:\heads" : "/var/heads";
+        var export = configure switch
+        {
+            "disabled-bad" => new AuditChainHeadExportOptions { Enabled = false, IntervalMinutes = 0 },
+            "file-ok" => new AuditChainHeadExportOptions { Enabled = true, File = { Directory = root } },
+            "file-relative" => new AuditChainHeadExportOptions { Enabled = true, File = { Directory = "heads" } },
+            "file-missing" => new AuditChainHeadExportOptions { Enabled = true },
+            "syslog-ok" => new AuditChainHeadExportOptions { Enabled = true, Sink = AuditChainHeadSinkKind.Syslog, Syslog = { Host = "collector" } },
+            "syslog-no-host" => new AuditChainHeadExportOptions { Enabled = true, Sink = AuditChainHeadSinkKind.Syslog },
+            "syslog-bad-port" => new AuditChainHeadExportOptions { Enabled = true, Sink = AuditChainHeadSinkKind.Syslog, Syslog = { Host = "c", Port = 0 } },
+            "syslog-bad-facility" => new AuditChainHeadExportOptions { Enabled = true, Sink = AuditChainHeadSinkKind.Syslog, Syslog = { Host = "c", Facility = 24 } },
+            _ => new AuditChainHeadExportOptions { Enabled = true, IntervalMinutes = 0, File = { Directory = root } },
+        };
+
+        var result = new AuditLogHashChainOptionsValidator().Validate(null, new AuditLogHashChainOptions { HeadExport = export });
+
+        if (expectedError is null)
+        {
+            Assert.True(result.Succeeded, result.FailureMessage);
+        }
+        else
+        {
+            Assert.True(result.Failed);
+            Assert.Contains(expectedError, result.FailureMessage);
+        }
+    }
+}
@@ -0,0 +1,135 @@
+using System.Net;
+using System.Net.Sockets;
+using System.Text;
+using System.Text.Json;
+using ZB.MOM.WW.ScadaBridge.AuditLog.Central.HeadExport;
+using ZB.MOM.WW.ScadaBridge.Commons.Types.Audit;
+
+namespace ZB.MOM.WW.ScadaBridge.AuditLog.Tests.Central.HeadExport;
+
+/// <summary>
+/// The two off-box chain-head sinks (ScadaBridge#107 item 1): the append-only JSON-lines file
+/// sink, and RFC 5424 syslog over UDP and TCP (octet-counted framing).
+/// </summary>
+public sealed class AuditChainHeadSinkTests : IDisposable
+{
+    private readonly DirectoryInfo _dir = Directory.CreateTempSubdirectory("head-sink-");
+
+    /// <inheritdoc />
+    public void Dispose() => _dir.Delete(recursive: true);
+
+    private static AuditChainHeadRecord Record(string month, long seq) => new(
+        1, month, seq, new string('a', 64), null, null, seq, 0,
+        new DateTime(2026, 5, 3, 0, 0, 0, DateTimeKind.Utc), new DateTime(2026, 5, 3, 1, 0, 0, DateTimeKind.Utc), "central-a");
+
+    private sealed class FixedClock(DateTimeOffset now) : TimeProvider
+    {
+        public override DateTimeOffset GetUtcNow() => now;
+    }
+
+    /// <summary>
+    /// Writes append to the node's file for the day — never truncate it — one JSON record per
+    /// line, and a pre-existing line survives.
+    /// </summary>
+    /// <returns>A task that represents the asynchronous operation.</returns>
+    [Fact]
+    public async Task FileSink_AppendsJsonLines_ToThePerNodePerDayFile_WithoutTruncating()
+    {
+        var clock = new FixedClock(new DateTimeOffset(2026, 5, 3, 1, 0, 0, TimeSpan.Zero));
+        var sink = new FileAuditChainHeadSink(_dir.FullName, "central-a", clock);
+        var path = sink.PathFor(clock.GetUtcNow().UtcDateTime);
+        Assert.Equal(Path.Combine(_dir.FullName, "2026", "audit-chain-heads-2026-05-03-central-a.jsonl"), path);
+        Directory.CreateDirectory(Path.GetDirectoryName(path)!);
+        await File.WriteAllTextAsync(path, "earlier line\n");
+
+        await sink.WriteAsync([Record("2026-04", 3), Record("2026-05", 7)], default);
+        await sink.WriteAsync([Record("2026-05", 9)], default);
+
+        var lines = await File.ReadAllLinesAsync(path);
+        Assert.Equal(4, lines.Length);
+        Assert.Equal("earlier line", lines[0]);
+        var parsed = lines.Skip(1).Select(l => JsonSerializer.Deserialize<AuditChainHeadRecord>(l, new JsonSerializerOptions(JsonSerializerDefaults.Web))!).ToList();
+        Assert.Equal([3L, 7L, 9L], parsed.Select(r => r.LastSeq));
+        Assert.Equal(Record("2026-05", 9), parsed[2]);
+    }
+
+    /// <summary>A relative export directory is refused.</summary>
+    [Fact]
+    public void FileSink_RejectsARelativeDirectory()
+    {
+        Assert.Throws<ArgumentException>(() => new FileAuditChainHeadSink("relative/dir", "central-a"));
+    }
+
+    /// <summary>UDP: one RFC 5424 datagram per record — PRI, version, MSGID, nil SD, JSON body.</summary>
+    /// <returns>A task that represents the asynchronous operation.</returns>
+    [Fact]
+    public async Task SyslogSink_Udp_SendsOneRfc5424DatagramPerRecord()
+    {
+        using var server = new UdpClient(new IPEndPoint(IPAddress.Loopback, 0));
+        var port = ((IPEndPoint)server.Client.LocalEndPoint!).Port;
+        var sink = new SyslogAuditChainHeadSink(
+            new AuditChainHeadSyslogSinkOptions { Host = "127.0.0.1", Port = port, Protocol = AuditSyslogProtocol.Udp, Facility = 13 },
+            hostName: "central a", time: new FixedClock(new DateTimeOffset(2026, 5, 3, 1, 2, 3, 456, TimeSpan.Zero)));
+
+        await sink.WriteAsync([Record("2026-04", 3), Record("2026-05", 7)], default);
+
+        var first = Encoding.UTF8.GetString((await server.ReceiveAsync().WaitAsync(TimeSpan.FromSeconds(5))).Buffer);
+        var second = Encoding.UTF8.GetString((await server.ReceiveAsync().WaitAsync(TimeSpan.FromSeconds(5))).Buffer);
+        // facility 13 (log audit) * 8 + severity 5 (notice) = 109; the space in the host name is dropped.
+        Assert.StartsWith($"<109>1 2026-05-03T01:02:03.456Z centrala scadabridge {Environment.ProcessId} chainhead - {{", first);
+        AssertRecordBody(first, 3);
+        AssertRecordBody(second, 7);
+    }
+
+    /// <summary>TCP: octet-counted frames (RFC 6587), one per record, on one connection.</summary>
+    /// <returns>A task that represents the asynchronous operation.</returns>
+    [Fact]
+    public async Task SyslogSink_Tcp_SendsOctetCountedFrames()
+    {
+        var listener = new TcpListener(IPAddress.Loopback, 0);
+        listener.Start();
+        try
+        {
+            var port = ((IPEndPoint)listener.LocalEndpoint).Port;
+            var received = Task.Run(async () =>
+            {
+                using var client = await listener.AcceptTcpClientAsync();
+                using var ms = new MemoryStream();
+                await client.GetStream().CopyToAsync(ms);
+                return ms.ToArray();
+            });
+            var sink = new SyslogAuditChainHeadSink(
+                new AuditChainHeadSyslogSinkOptions { Host = "127.0.0.1", Port = port, Protocol = AuditSyslogProtocol.Tcp });
+
+            await sink.WriteAsync([Record("2026-04", 3), Record("2026-05", 7)], default);
+
+            var bytes = await received.WaitAsync(TimeSpan.FromSeconds(5));
+            var frames = new List<string>();
+            var i = 0;
+            while (i < bytes.Length)
+            {
+                var space = Array.IndexOf(bytes, (byte)' ', i);
+                var length = int.Parse(Encoding.ASCII.GetString(bytes, i, space - i));
+                frames.Add(Encoding.UTF8.GetString(bytes, space + 1, length));
+                i = space + 1 + length;
+            }
+
+            Assert.Equal(2, frames.Count);
+            Assert.All(frames, f => Assert.StartsWith("<109>1 ", f));
+            AssertRecordBody(frames[0], 3);
+            AssertRecordBody(frames[1], 7);
+        }
+        finally
+        {
+            listener.Stop();
+        }
+    }
+
+    private static void AssertRecordBody(string message, long seq)
+    {
+        var json = message[message.IndexOf('{')..];
+        var record = JsonSerializer.Deserialize<AuditChainHeadRecord>(json, new JsonSerializerOptions(JsonSerializerDefaults.Web))!;
+        Assert.Equal(seq, record.LastSeq);
+        Assert.Equal(new string('a', 64), record.LastHash);
+    }
+}