From d1fc336e18288fe71f7cde7d5ff3b06ee5ce91b7 Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Mon, 9 Nov 2026 17:00:58 -0500 Subject: [PATCH] feat(management): parquet audit export and audit archive list/verify (ScadaBridge#42) GET /api/audit/export?format=parquet returned 501. It now streams the same 24 columns as csv/jsonl (AuditExportRow, CsvHeader order) as one Parquet file, one 10,000-row row group at a time. Each row group is encoded into a small non-seekable staging buffer and copied to the response asynchronously, because Kestrel rejects the synchronous writes the Parquet writer issues. New management commands, with CLI parity: - ListAuditArchives -> `audit archive list` (AuditArchiveListResult: enabled flag + every manifest). - VerifyAuditArchive(Name) -> `audit archive verify --name` (files, chain, purge anchor); exit 0 verified, 3 not verified, 1 error, 2 unauthorized. Both carry verify-chain's gate: OperationalAudit roles (Administrator, Viewer), site-scoped non-Administrator refused because an archive spans every site. VerifyAuditArchive uses the long-running Ask timeout. Error codes ARCHIVE_NOT_CONFIGURED, INVALID_ARGUMENT, NOT_FOUND. RequiredRoleMatrixTests freezes both gates; ManagementSurfaceParityTests now probes `audit archive list|verify` alongside `audit verify-chain`. Claude-Session: https://claude.ai/code/session_01GT67AEHd2Acv1YPwGfXEoV --- .../Commands/AuditArchiveHelpers.cs | 117 ++++++++++ .../Commands/AuditCommands.cs | 48 +++- .../Commands/AuditExportHelpers.cs | 4 +- .../Messages/Management/AuditLogCommands.cs | 16 ++ .../AuditEndpoints.cs | 56 +++-- .../AuditExportParquet.cs | 219 ++++++++++++++++++ .../ManagementActor.cs | 60 +++++ .../ManagementEndpoints.cs | 3 + ...OM.WW.ScadaBridge.ManagementService.csproj | 1 + .../CommandTreeTests.cs | 3 +- .../Commands/AuditArchiveCommandTests.cs | 136 +++++++++++ .../Commands/AuditCommandsScaffoldTests.cs | 11 +- .../ManagementSurfaceParityTests.cs | 10 +- .../AuditArchiveCommandTests.cs | 147 ++++++++++++ .../AuditEndpointsTests.cs | 40 +++- .../AuditExportParquetTests.cs | 150 ++++++++++++ .../RequiredRoleMatrixTests.cs | 3 + 17 files changed, 982 insertions(+), 42 deletions(-) create mode 100644 src/ZB.MOM.WW.ScadaBridge.CLI/Commands/AuditArchiveHelpers.cs create mode 100644 src/ZB.MOM.WW.ScadaBridge.ManagementService/AuditExportParquet.cs create mode 100644 tests/ZB.MOM.WW.ScadaBridge.CLI.Tests/Commands/AuditArchiveCommandTests.cs create mode 100644 tests/ZB.MOM.WW.ScadaBridge.ManagementService.Tests/AuditArchiveCommandTests.cs create mode 100644 tests/ZB.MOM.WW.ScadaBridge.ManagementService.Tests/AuditExportParquetTests.cs diff --git a/src/ZB.MOM.WW.ScadaBridge.CLI/Commands/AuditArchiveHelpers.cs b/src/ZB.MOM.WW.ScadaBridge.CLI/Commands/AuditArchiveHelpers.cs new file mode 100644 index 00000000..20f053ef --- /dev/null +++ b/src/ZB.MOM.WW.ScadaBridge.CLI/Commands/AuditArchiveHelpers.cs @@ -0,0 +1,117 @@ +using System.Globalization; +using System.Text.Json; + +namespace ZB.MOM.WW.ScadaBridge.CLI.Commands; + +/// +/// Helpers for the audit archive list / audit archive verify subcommands +/// (AuditLog Parquet cold archive, ScadaBridge#42). Both send a management command +/// (ListAuditArchives, VerifyAuditArchive); the archives live on the central +/// nodes' archive directory, not on the operator's machine. +/// +/// +/// verify exit codes: 0 verified, when the +/// files, the chain or the purge anchor disagree, 1 on a request error, 2 on an +/// authorization failure. +/// +public static class AuditArchiveHelpers +{ + /// Process exit code when an archive failed verification (tamper evidence). + public const int NotVerifiedExitCode = 3; + + /// + /// Client deadline for verify: the server hashes and re-chains a whole month under + /// the long-running management Ask (5 min); the client waits a little longer. + /// + public static readonly TimeSpan VerifyRequestTimeout = TimeSpan.FromMinutes(6); + + /// Renders a ListAuditArchives result. + /// The success body (an AuditArchiveListResult, camelCase). + /// Resolved output format. + /// Writer for the report. + /// Always 0. + public static int RenderList(string json, string format, TextWriter output) + { + if (!string.Equals(format, "table", StringComparison.OrdinalIgnoreCase)) + { + output.WriteLine(json); + output.Flush(); + return 0; + } + + using var doc = JsonDocument.Parse(json); + var root = doc.RootElement; + var enabled = root.TryGetProperty("enabled", out var e) && e.ValueKind == JsonValueKind.True; + output.WriteLine($"Archiving: {(enabled ? "enabled" : "disabled")}"); + if (!root.TryGetProperty("archives", out var archives) || archives.GetArrayLength() == 0) + { + output.WriteLine("No archives."); + output.Flush(); + return 0; + } + + output.WriteLine($"{"NAME",-42} {"PARTITION",-9} {"STATUS",-8} {"ROWS",12} {"CHAINED",12} {"BYTES",14}"); + foreach (var a in archives.EnumerateArray()) + { + output.WriteLine(string.Create(CultureInfo.InvariantCulture, + $"{Str(a, "name"),-42} {Str(a, "partition"),-9} {Str(a, "status"),-8} {Num(a, "rowCount"),12} {Num(a, "chainedRowCount"),12} {Num(a, "dataBytes"),14}")); + } + + output.Flush(); + return 0; + } + + /// Renders a VerifyAuditArchive result and maps it to the exit code. + /// The success body (an AuditArchiveVerificationResult, camelCase). + /// Resolved output format. + /// Writer for the report. + /// 0 when verified, otherwise. + public static int RenderVerify(string json, string format, TextWriter output) + { + using var doc = JsonDocument.Parse(json); + var root = doc.RootElement; + var verified = root.TryGetProperty("verified", out var v) && v.ValueKind == JsonValueKind.True; + + if (!string.Equals(format, "table", StringComparison.OrdinalIgnoreCase)) + { + output.WriteLine(json); + output.Flush(); + return verified ? 0 : NotVerifiedExitCode; + } + + output.WriteLine($"Archive : {Str(root, "name")}"); + output.WriteLine($"Partition : {Str(root, "partition")}"); + output.WriteLine($"Manifest : {Str(root, "manifestStatus")}"); + output.WriteLine($"Verified : {(verified ? "yes" : "NO")}"); + output.WriteLine($"Files intact: {(root.TryGetProperty("filesIntact", out var f) && f.ValueKind == JsonValueKind.True ? "yes" : "NO")}"); + if (Str(root, "fileProblem") is { Length: > 0 } problem) + { + output.WriteLine($" problem : {problem}"); + } + + output.WriteLine($"Rows read : {Num(root, "rowsRead")}"); + if (root.TryGetProperty("chains", out var chains) && chains.ValueKind == JsonValueKind.Array) + { + foreach (var c in chains.EnumerateArray()) + { + var chain = c.GetProperty("chain"); + output.WriteLine($"Chain {Str(chain, "partition")}: {Str(chain, "status")}, {Num(chain, "rowsVerified")} rows verified, " + + $"{Num(chain, "tombstonesApplied")} tombstones, {Num(chain, "unchainedRows")} unchained; anchor {Str(c, "anchor")}"); + output.WriteLine($" {Str(c, "anchorDetail")}"); + if (chain.TryGetProperty("firstBreak", out var brk) && brk.ValueKind == JsonValueKind.Object) + { + output.WriteLine($" first break: chainSeq {Num(brk, "chainSeq")}, eventId {Str(brk, "eventId", "-")}, {Str(brk, "kind")} — {Str(brk, "detail")}"); + } + } + } + + output.Flush(); + return verified ? 0 : NotVerifiedExitCode; + } + + private static string Str(JsonElement e, string name, string missing = "") + => e.TryGetProperty(name, out var v) && v.ValueKind == JsonValueKind.String ? v.GetString() ?? missing : missing; + + private static string Num(JsonElement e, string name, string missing = "0") + => e.TryGetProperty(name, out var v) && v.ValueKind == JsonValueKind.Number ? v.GetInt64().ToString(CultureInfo.InvariantCulture) : missing; +} diff --git a/src/ZB.MOM.WW.ScadaBridge.CLI/Commands/AuditCommands.cs b/src/ZB.MOM.WW.ScadaBridge.CLI/Commands/AuditCommands.cs index 8ec96f77..91976085 100644 --- a/src/ZB.MOM.WW.ScadaBridge.CLI/Commands/AuditCommands.cs +++ b/src/ZB.MOM.WW.ScadaBridge.CLI/Commands/AuditCommands.cs @@ -9,7 +9,9 @@ namespace ZB.MOM.WW.ScadaBridge.CLI.Commands; /// the centralized append-only Audit Log via the REST endpoints /// (GET /api/audit/query, GET /api/audit/export, /// GET /api/audit/tree), plus verify-chain, which sends the -/// VerifyAuditChain management command (hash-chain tamper evidence, ScadaBridge#41). +/// VerifyAuditChain management command (hash-chain tamper evidence, ScadaBridge#41), +/// and archive list / archive verify, which send ListAuditArchives / +/// VerifyAuditArchive (Parquet cold archive, ScadaBridge#42). /// public static class AuditCommands { @@ -31,6 +33,7 @@ public static class AuditCommands command.Add(BuildTree(urlOption, formatOption, usernameOption, passwordOption)); command.Add(BuildVerifyChain(urlOption, formatOption, usernameOption, passwordOption)); command.Add(BuildBackfillSourceNode(urlOption, formatOption, usernameOption, passwordOption)); + command.Add(BuildArchive(urlOption, formatOption, usernameOption, passwordOption)); return command; } @@ -316,6 +319,49 @@ public static class AuditCommands return cmd; } + /// + /// Builds the audit archive group (ScadaBridge#42): list and verify + /// over the AuditLog Parquet cold archives on the central archive directory. + /// + private static Command BuildArchive(Option urlOption, Option formatOption, Option usernameOption, Option passwordOption) + { + var group = new Command("archive") { Description = "List and verify the AuditLog Parquet cold archives" }; + + var list = new Command("list") { Description = "List the AuditLog Parquet archives written by the retention purge" }; + list.SetAction(async (ParseResult result) => + { + var format = AuditCommandHelpers.ResolveFormat(result, formatOption); + return await CommandHelpers.ExecuteCommandAsync( + result, urlOption, formatOption, usernameOption, passwordOption, + new ListAuditArchivesCommand(), + onSuccess: json => AuditArchiveHelpers.RenderList(json, format, Console.Out)); + }); + group.Add(list); + + var nameOption = new Option("--name") + { + Description = "Archive name, as 'audit archive list' prints it (auditlog-YYYY-MM-yyyyMMddTHHmmssfffZ)", + Required = true, + }; + var verify = new Command("verify") + { + Description = "Verify an archive's files, its hash chain and its purge anchor; exit 3 if it does not verify", + }; + verify.Add(nameOption); + verify.SetAction(async (ParseResult result) => + { + var format = AuditCommandHelpers.ResolveFormat(result, formatOption); + return await CommandHelpers.ExecuteCommandAsync( + result, urlOption, formatOption, usernameOption, passwordOption, + new VerifyAuditArchiveCommand(result.GetValue(nameOption)!), + timeout: AuditArchiveHelpers.VerifyRequestTimeout, + onSuccess: json => AuditArchiveHelpers.RenderVerify(json, format, Console.Out)); + }); + group.Add(verify); + + return group; + } + /// /// Builds the audit backfill-source-node sub-command. /// Sets SourceNode on historical pre-feature rows whose SourceNode IS NULL diff --git a/src/ZB.MOM.WW.ScadaBridge.CLI/Commands/AuditExportHelpers.cs b/src/ZB.MOM.WW.ScadaBridge.CLI/Commands/AuditExportHelpers.cs index b57a82c7..3a5e46e3 100644 --- a/src/ZB.MOM.WW.ScadaBridge.CLI/Commands/AuditExportHelpers.cs +++ b/src/ZB.MOM.WW.ScadaBridge.CLI/Commands/AuditExportHelpers.cs @@ -134,8 +134,8 @@ public static class AuditExportHelpers /// Executes the export: GETs /api/audit/export and copies the response body /// stream directly to . The body is never fully /// buffered — the copy streams in fixed-size chunks via . - /// A 501 Not Implemented (parquet not yet supported server-side) prints the - /// server message and returns a non-zero exit code. + /// A 501 Not Implemented (a central older than ScadaBridge#42 does not serve + /// format=parquet) prints the server message and returns a non-zero exit code. /// /// The management HTTP client for API communication. /// The export arguments containing filters and output file path. diff --git a/src/ZB.MOM.WW.ScadaBridge.Commons/Messages/Management/AuditLogCommands.cs b/src/ZB.MOM.WW.ScadaBridge.Commons/Messages/Management/AuditLogCommands.cs index 8a024ae2..5fd94e17 100644 --- a/src/ZB.MOM.WW.ScadaBridge.Commons/Messages/Management/AuditLogCommands.cs +++ b/src/ZB.MOM.WW.ScadaBridge.Commons/Messages/Management/AuditLogCommands.cs @@ -31,3 +31,19 @@ public record AuditLogQueryResult(IReadOnlyList Entries, int Tota /// /// The month to verify, YYYY-MM. public record VerifyAuditChainCommand(string Month); + +/// +/// Lists the AuditLog Parquet cold archives under the configured archive directory (audit +/// roadmap T2, ScadaBridge#42). Read-only; OperationalAudit roles, refused for a site-scoped +/// caller because an archive holds every site's rows. +/// +public record ListAuditArchivesCommand; + +/// +/// Verifies one AuditLog Parquet archive (ScadaBridge#42): file integrity against its +/// manifest, the hash chain recomputed over the archived rows, and the chain end against the +/// checkpoint anchor the partition purge left. Read-only; same gate as +/// . +/// +/// The archive name, as audit archive list reports it. +public record VerifyAuditArchiveCommand(string Name); diff --git a/src/ZB.MOM.WW.ScadaBridge.ManagementService/AuditEndpoints.cs b/src/ZB.MOM.WW.ScadaBridge.ManagementService/AuditEndpoints.cs index 1c5f983b..01c44dce 100644 --- a/src/ZB.MOM.WW.ScadaBridge.ManagementService/AuditEndpoints.cs +++ b/src/ZB.MOM.WW.ScadaBridge.ManagementService/AuditEndpoints.cs @@ -22,8 +22,8 @@ namespace ZB.MOM.WW.ScadaBridge.ManagementService; /// /// GET /api/audit/query — keyset-paged JSON page, gated on the /// permission. -/// GET /api/audit/export — streamed bulk export (csv / jsonl; -/// parquet returns HTTP 501), gated on the +/// GET /api/audit/export — streamed bulk export (csv / jsonl / +/// parquet), gated on the /// permission. /// GET /api/audit/tree — execution-chain tree rooted at the /// topmost ancestor of a given executionId, returned as a JSON array @@ -47,11 +47,8 @@ namespace ZB.MOM.WW.ScadaBridge.ManagementService; /// /// /// -/// Parquet. No Parquet library is referenced by the solution and -/// Component-AuditLog.md explicitly defers Parquet archival to v1.x, so -/// format=parquet returns 501 Not Implemented with a guidance -/// message rather than a half-built binary stream. csv + jsonl are fully -/// implemented. +/// Parquet. format=parquet (ScadaBridge#42) streams the same 24 columns as +/// csv / jsonl as one Parquet file (), a row group at a time. /// /// public static class AuditEndpoints @@ -65,6 +62,9 @@ public static class AuditEndpoints /// Repository round-trip size used while streaming an export. public const int ExportPageSize = 1000; + /// Content type of a format=parquet export (ScadaBridge#42). + public const string ParquetContentType = "application/vnd.apache.parquet"; + /// /// Serializer for individual rows (query results + /// jsonl export lines). Drops null columns so a sparse audit row stays @@ -185,10 +185,10 @@ public static class AuditEndpoints // ───────────────────────────────────────────────────────────────────── /// - /// Handles GET /api/audit/export: authenticates, checks the AuditExport permission, and streams the matching rows as CSV or JSONL. + /// Handles GET /api/audit/export: authenticates, checks the AuditExport permission, and streams the matching rows as CSV, JSONL or Parquet. /// /// The HTTP context for the current request. - /// A task that resolves to the HTTP result (streaming CSV/JSONL, 400, 401, 403, or 501). + /// A task that resolves to the HTTP result (streaming CSV/JSONL/Parquet, 400, 401, or 403). internal static async Task HandleExport(HttpContext context) { var auth = await AuthenticateAsync(context); @@ -208,17 +208,7 @@ public static class AuditEndpoints format = "csv"; } - if (format == "parquet") - { - // Deferred to v1.x (Component-AuditLog.md §"Deferred"): no Parquet - // library is in the solution. Return a 501 with actionable guidance - // rather than a partial binary stream. - return Results.Json( - new { error = "Parquet export deferred to v1.x; use csv or jsonl.", code = "NOT_IMPLEMENTED" }, - statusCode: 501); - } - - if (format != "csv" && format != "jsonl") + if (format != "csv" && format != "jsonl" && format != "parquet") { return Results.Json( new { error = $"Unsupported export format '{format}'. Use csv, jsonl, or parquet.", code = "BAD_REQUEST" }, @@ -243,8 +233,12 @@ public static class AuditEndpoints filter = restricted; var repo = context.RequestServices.GetRequiredService(); - var contentType = format == "csv" ? "text/csv; charset=utf-8" : "application/x-ndjson"; - var extension = format == "csv" ? "csv" : "jsonl"; + var (contentType, extension) = format switch + { + "csv" => ("text/csv; charset=utf-8", "csv"), + "parquet" => (ParquetContentType, "parquet"), + _ => ("application/x-ndjson", "jsonl"), + }; var fileName = $"audit-log-{DateTime.UtcNow:yyyyMMdd-HHmmss}.{extension}"; context.Response.ContentType = contentType; @@ -256,6 +250,12 @@ public static class AuditEndpoints { await StreamCsvAsync(repo, filter, context.Response.Body, context.RequestAborted); } + else if (format == "parquet") + { + // ScadaBridge#42: same 24 columns as csv/jsonl, one row group per 10k rows. + await AuditExportParquet.WriteAsync( + ExportRowPagesAsync(repo, filter, context.RequestAborted), context.Response.Body, context.RequestAborted); + } else { await StreamJsonlAsync(repo, filter, context.Response.Body, context.RequestAborted); @@ -494,6 +494,18 @@ public static class AuditEndpoints } } + /// Projects pages onto the flat export row shape. + private static async IAsyncEnumerable> ExportRowPagesAsync( + IAuditLogRepository repo, + AuditLogQueryFilter filter, + [System.Runtime.CompilerServices.EnumeratorCancellation] CancellationToken ct) + { + await foreach (var page in PageAsync(repo, filter, ct)) + { + yield return page.Select(AuditExportRow.From).ToList(); + } + } + /// /// Lazily yields full repository pages, advancing the keyset cursor until a /// short page signals the end. Shared by the csv + jsonl streamers. diff --git a/src/ZB.MOM.WW.ScadaBridge.ManagementService/AuditExportParquet.cs b/src/ZB.MOM.WW.ScadaBridge.ManagementService/AuditExportParquet.cs new file mode 100644 index 00000000..1a829b82 --- /dev/null +++ b/src/ZB.MOM.WW.ScadaBridge.ManagementService/AuditExportParquet.cs @@ -0,0 +1,219 @@ +using Parquet; +using Parquet.Schema; + +namespace ZB.MOM.WW.ScadaBridge.ManagementService; + +/// +/// GET /api/audit/export?format=parquet (ScadaBridge#42): the same 24 columns as the +/// csv / jsonl export (, in +/// order) as one Parquet file, one row group per rows. +/// +/// +/// +/// Types: ids are D-format strings, timestamps are UTC Parquet timestamps +/// (microseconds), enums are their names, HttpStatus/DurationMs are int32, +/// PayloadTruncated is boolean, everything else is a UTF-8 string. +/// +/// +/// Streaming. Kestrel forbids synchronous writes to the response body and the Parquet +/// writer issues some, so each row group is encoded into a small in-memory buffer and then +/// copied to the response asynchronously. Memory is bounded by one row group, and the client +/// starts receiving bytes after the first one. A Parquet file's footer is written last, so a +/// response cut off mid-stream is not a readable file — the CLI treats it as a failed export. +/// +/// +/// This is the ad-hoc export. The retention archive (alog.md §15.2) is a different layout +/// that carries the hash-chain columns; see AuditArchiveParquet. +/// +/// +public static class AuditExportParquet +{ + /// Rows per Parquet row group. + public const int RowGroupRows = 10_000; + + private static readonly DataField EventId = new(nameof(AuditExportRow.EventId), false); + private static readonly DateTimeDataField OccurredAtUtc = Timestamp(nameof(AuditExportRow.OccurredAtUtc), false); + private static readonly DateTimeDataField IngestedAtUtc = Timestamp(nameof(AuditExportRow.IngestedAtUtc), true); + private static readonly DataField Channel = new(nameof(AuditExportRow.Channel), false); + private static readonly DataField Kind = new(nameof(AuditExportRow.Kind), false); + private static readonly DataField CorrelationId = Text(nameof(AuditExportRow.CorrelationId)); + private static readonly DataField ExecutionId = Text(nameof(AuditExportRow.ExecutionId)); + private static readonly DataField ParentExecutionId = Text(nameof(AuditExportRow.ParentExecutionId)); + private static readonly DataField SourceSiteId = Text(nameof(AuditExportRow.SourceSiteId)); + private static readonly DataField SourceNode = Text(nameof(AuditExportRow.SourceNode)); + private static readonly DataField SourceInstanceId = Text(nameof(AuditExportRow.SourceInstanceId)); + private static readonly DataField SourceScript = Text(nameof(AuditExportRow.SourceScript)); + private static readonly DataField Actor = Text(nameof(AuditExportRow.Actor)); + private static readonly DataField Target = Text(nameof(AuditExportRow.Target)); + private static readonly DataField Status = new(nameof(AuditExportRow.Status), false); + private static readonly DataField HttpStatus = new(nameof(AuditExportRow.HttpStatus), true); + private static readonly DataField DurationMs = new(nameof(AuditExportRow.DurationMs), true); + private static readonly DataField ErrorMessage = Text(nameof(AuditExportRow.ErrorMessage)); + private static readonly DataField ErrorDetail = Text(nameof(AuditExportRow.ErrorDetail)); + private static readonly DataField RequestSummary = Text(nameof(AuditExportRow.RequestSummary)); + private static readonly DataField ResponseSummary = Text(nameof(AuditExportRow.ResponseSummary)); + private static readonly DataField PayloadTruncated = new(nameof(AuditExportRow.PayloadTruncated), false); + private static readonly DataField Extra = Text(nameof(AuditExportRow.Extra)); + private static readonly DataField ForwardState = Text(nameof(AuditExportRow.ForwardState)); + + /// The export schema; column names and order equal . + public static ParquetSchema Schema { get; } = new( + EventId, OccurredAtUtc, IngestedAtUtc, Channel, Kind, CorrelationId, ExecutionId, ParentExecutionId, + SourceSiteId, SourceNode, SourceInstanceId, SourceScript, Actor, Target, Status, HttpStatus, DurationMs, + ErrorMessage, ErrorDetail, RequestSummary, ResponseSummary, PayloadTruncated, Extra, ForwardState); + + /// Writes every row of to as one Parquet file. + /// Export rows, page by page. + /// The destination (e.g. the HTTP response body); written asynchronously only. + /// Cancellation token. + /// The number of rows written. + public static async Task WriteAsync( + IAsyncEnumerable> pages, Stream output, CancellationToken ct = default) + { + ArgumentNullException.ThrowIfNull(pages); + ArgumentNullException.ThrowIfNull(output); + + // Non-seekable on purpose: the writer then tracks file offsets itself, so emptying the + // buffer after each drain cannot corrupt the column-chunk offsets in the footer. + var staging = new StagingStream(); + long total = 0; + var buffer = new List(RowGroupRows); + await using (var writer = await ParquetWriter.CreateAsync(Schema, staging, cancellationToken: ct).ConfigureAwait(false)) + { + await foreach (var page in pages.WithCancellation(ct).ConfigureAwait(false)) + { + foreach (var row in page) + { + buffer.Add(row); + if (buffer.Count == RowGroupRows) + { + total += await WriteGroupAsync(writer, buffer, ct).ConfigureAwait(false); + await DrainAsync(staging, output, ct).ConfigureAwait(false); + } + } + } + + total += await WriteGroupAsync(writer, buffer, ct).ConfigureAwait(false); + } + + // The footer is written when the writer is disposed. + await DrainAsync(staging, output, ct).ConfigureAwait(false); + return total; + } + + private static async Task WriteGroupAsync(ParquetWriter writer, List rows, CancellationToken ct) + { + if (rows.Count == 0) + { + return 0; + } + + using (var group = writer.CreateRowGroup()) + { + await group.WriteAsync(EventId, Map(rows, r => r.EventId.ToString("D"))).ConfigureAwait(false); + await group.WriteAsync(OccurredAtUtc, rows.Select(r => DateTime.SpecifyKind(r.OccurredAtUtc, DateTimeKind.Utc)).ToArray(), cancellationToken: ct).ConfigureAwait(false); + await group.WriteAsync(IngestedAtUtc, rows.Select(r => r.IngestedAtUtc is { } i ? DateTime.SpecifyKind(i, DateTimeKind.Utc) : (DateTime?)null).ToArray(), cancellationToken: ct).ConfigureAwait(false); + await group.WriteAsync(Channel, Map(rows, r => r.Channel.ToString())).ConfigureAwait(false); + await group.WriteAsync(Kind, Map(rows, r => r.Kind.ToString())).ConfigureAwait(false); + await group.WriteAsync(CorrelationId, Map(rows, r => r.CorrelationId?.ToString("D"))!).ConfigureAwait(false); + await group.WriteAsync(ExecutionId, Map(rows, r => r.ExecutionId?.ToString("D"))!).ConfigureAwait(false); + await group.WriteAsync(ParentExecutionId, Map(rows, r => r.ParentExecutionId?.ToString("D"))!).ConfigureAwait(false); + await group.WriteAsync(SourceSiteId, Map(rows, r => r.SourceSiteId)!).ConfigureAwait(false); + await group.WriteAsync(SourceNode, Map(rows, r => r.SourceNode)!).ConfigureAwait(false); + await group.WriteAsync(SourceInstanceId, Map(rows, r => r.SourceInstanceId)!).ConfigureAwait(false); + await group.WriteAsync(SourceScript, Map(rows, r => r.SourceScript)!).ConfigureAwait(false); + await group.WriteAsync(Actor, Map(rows, r => r.Actor)!).ConfigureAwait(false); + await group.WriteAsync(Target, Map(rows, r => r.Target)!).ConfigureAwait(false); + await group.WriteAsync(Status, Map(rows, r => r.Status.ToString())).ConfigureAwait(false); + await group.WriteAsync(HttpStatus, rows.Select(r => r.HttpStatus).ToArray(), cancellationToken: ct).ConfigureAwait(false); + await group.WriteAsync(DurationMs, rows.Select(r => r.DurationMs).ToArray(), cancellationToken: ct).ConfigureAwait(false); + await group.WriteAsync(ErrorMessage, Map(rows, r => r.ErrorMessage)!).ConfigureAwait(false); + await group.WriteAsync(ErrorDetail, Map(rows, r => r.ErrorDetail)!).ConfigureAwait(false); + await group.WriteAsync(RequestSummary, Map(rows, r => r.RequestSummary)!).ConfigureAwait(false); + await group.WriteAsync(ResponseSummary, Map(rows, r => r.ResponseSummary)!).ConfigureAwait(false); + await group.WriteAsync(PayloadTruncated, rows.Select(r => r.PayloadTruncated).ToArray(), cancellationToken: ct).ConfigureAwait(false); + await group.WriteAsync(Extra, Map(rows, r => r.Extra)!).ConfigureAwait(false); + await group.WriteAsync(ForwardState, Map(rows, r => r.ForwardState?.ToString())!).ConfigureAwait(false); + } + + var count = rows.Count; + rows.Clear(); + return count; + } + + private static async Task DrainAsync(StagingStream staging, Stream output, CancellationToken ct) + { + var pending = staging.Pending; + if (pending.Length == 0) + { + return; + } + + await output.WriteAsync(pending, ct).ConfigureAwait(false); + await output.FlushAsync(ct).ConfigureAwait(false); + staging.Clear(); + } + + private static string?[] Map(List rows, Func select) + => rows.Select(select).ToArray(); + + private static DataField Text(string name) => new(name, true); + + private static DateTimeDataField Timestamp(string name, bool nullable) + => new(name, DateTimeFormat.Timestamp, isAdjustedToUTC: true, unit: DateTimeTimeUnit.Micros, isNullable: nullable); + + /// + /// Forward-only, non-seekable buffer the Parquet writer writes into (synchronously or + /// not); moves its bytes to the real output asynchronously. + /// + private sealed class StagingStream : Stream + { + private readonly MemoryStream _buffer = new(); + + public ReadOnlyMemory Pending => _buffer.GetBuffer().AsMemory(0, (int)_buffer.Length); + + public override bool CanRead => false; + + public override bool CanSeek => false; + + public override bool CanWrite => true; + + public override long Length => throw new NotSupportedException(); + + public override long Position + { + get => throw new NotSupportedException(); + set => throw new NotSupportedException(); + } + + public void Clear() => _buffer.SetLength(0); + + public override void Write(byte[] buffer, int offset, int count) => _buffer.Write(buffer, offset, count); + + public override void Write(ReadOnlySpan buffer) => _buffer.Write(buffer); + + public override ValueTask WriteAsync(ReadOnlyMemory buffer, CancellationToken cancellationToken = default) + { + _buffer.Write(buffer.Span); + return ValueTask.CompletedTask; + } + + public override Task WriteAsync(byte[] buffer, int offset, int count, CancellationToken cancellationToken) + { + _buffer.Write(buffer, offset, count); + return Task.CompletedTask; + } + + public override void Flush() + { + } + + public override Task FlushAsync(CancellationToken cancellationToken) => Task.CompletedTask; + + public override int Read(byte[] buffer, int offset, int count) => throw new NotSupportedException(); + + public override long Seek(long offset, SeekOrigin origin) => throw new NotSupportedException(); + + public override void SetLength(long value) => throw new NotSupportedException(); + } +} diff --git a/src/ZB.MOM.WW.ScadaBridge.ManagementService/ManagementActor.cs b/src/ZB.MOM.WW.ScadaBridge.ManagementService/ManagementActor.cs index cafd61c4..d4175567 100644 --- a/src/ZB.MOM.WW.ScadaBridge.ManagementService/ManagementActor.cs +++ b/src/ZB.MOM.WW.ScadaBridge.ManagementService/ManagementActor.cs @@ -393,6 +393,11 @@ public class ManagementActor : ReceiveActor // spans every site. VerifyAuditChainCommand => OperationalAuditReaders, + // AuditLog Parquet archives (ScadaBridge#42): audit reads with verify-chain's gate — + // OperationalAudit roles, site-scoped callers refused in the handler. + ListAuditArchivesCommand => OperationalAuditReaders, + VerifyAuditArchiveCommand => OperationalAuditReaders, + // Read-only queries -- any authenticated user _ => null }; @@ -564,6 +569,8 @@ public class ManagementActor : ReceiveActor // Audit Log QueryAuditLogCommand cmd => await HandleQueryAuditLog(sp, cmd), VerifyAuditChainCommand cmd => await HandleVerifyAuditChain(sp, cmd, user), + ListAuditArchivesCommand => await HandleListAuditArchives(sp, user), + VerifyAuditArchiveCommand cmd => await HandleVerifyAuditArchive(sp, cmd, user), // Health GetHealthSummaryCommand => HandleGetHealthSummary(sp), @@ -3256,6 +3263,59 @@ public class ManagementActor : ReceiveActor } } + /// + /// Lists the AuditLog Parquet archives (ScadaBridge#42). Refused for a site-scoped, + /// non-Administrator caller: an archive holds every site's rows. + /// + private static async Task HandleListAuditArchives(IServiceProvider sp, AuthenticatedUser user) + { + var archive = RequireAuditArchive(sp, user); + return new AuditArchiveListResult(archive.IsEnabled, await archive.ListAsync()); + } + + /// + /// Verifies one AuditLog Parquet archive (ScadaBridge#42): files against the manifest, + /// the chain recomputed from the archived rows, the chain end against the purge anchor. + /// + private static async Task HandleVerifyAuditArchive( + IServiceProvider sp, VerifyAuditArchiveCommand cmd, AuthenticatedUser user) + { + var archive = RequireAuditArchive(sp, user); + try + { + return await archive.VerifyAsync(cmd.Name); + } + catch (ArgumentException ex) + { + throw new ManagementCommandException(ex.Message, "INVALID_ARGUMENT"); + } + catch (FileNotFoundException ex) + { + throw new ManagementCommandException(ex.Message, "NOT_FOUND"); + } + } + + private static IAuditLogArchive RequireAuditArchive(IServiceProvider sp, AuthenticatedUser user) + { + if (user.PermittedSiteIds.Length > 0 + && !user.Roles.Contains(Roles.Administrator, StringComparer.OrdinalIgnoreCase)) + { + throw new SiteScopeViolationException( + "Access denied: AuditLog archives hold every site's rows; they need an " + + "OperationalAudit role that is not scoped to specific sites."); + } + + var archive = sp.GetService(); + if (archive is null || !archive.IsConfigured) + { + throw new ManagementCommandException( + "No AuditLog archive directory is configured on this central node (AuditLog:Archive:Directory).", + "ARCHIVE_NOT_CONFIGURED"); + } + + return archive; + } + // ======================================================================== // Health handlers // ======================================================================== diff --git a/src/ZB.MOM.WW.ScadaBridge.ManagementService/ManagementEndpoints.cs b/src/ZB.MOM.WW.ScadaBridge.ManagementService/ManagementEndpoints.cs index f0e12098..5e178854 100644 --- a/src/ZB.MOM.WW.ScadaBridge.ManagementService/ManagementEndpoints.cs +++ b/src/ZB.MOM.WW.ScadaBridge.ManagementService/ManagementEndpoints.cs @@ -63,6 +63,9 @@ public static class ManagementEndpoints // AuditLog hash-chain verification (ScadaBridge#41) recomputes one hash per row of // a whole month — millions of rows on a busy central. typeof(VerifyAuditChainCommand), + // AuditLog archive verification (ScadaBridge#42) hashes and re-chains a whole month's + // Parquet archive. + typeof(VerifyAuditArchiveCommand), ]; /// diff --git a/src/ZB.MOM.WW.ScadaBridge.ManagementService/ZB.MOM.WW.ScadaBridge.ManagementService.csproj b/src/ZB.MOM.WW.ScadaBridge.ManagementService/ZB.MOM.WW.ScadaBridge.ManagementService.csproj index 67db2fc2..4f76e22e 100644 --- a/src/ZB.MOM.WW.ScadaBridge.ManagementService/ZB.MOM.WW.ScadaBridge.ManagementService.csproj +++ b/src/ZB.MOM.WW.ScadaBridge.ManagementService/ZB.MOM.WW.ScadaBridge.ManagementService.csproj @@ -12,6 +12,7 @@ + diff --git a/tests/ZB.MOM.WW.ScadaBridge.CLI.Tests/CommandTreeTests.cs b/tests/ZB.MOM.WW.ScadaBridge.CLI.Tests/CommandTreeTests.cs index c185c1ca..0d17ebc8 100644 --- a/tests/ZB.MOM.WW.ScadaBridge.CLI.Tests/CommandTreeTests.cs +++ b/tests/ZB.MOM.WW.ScadaBridge.CLI.Tests/CommandTreeTests.cs @@ -69,7 +69,7 @@ public class CommandTreeTests Assert.Contains("bundle", groupNames); } - /// Verifies that the audit command group exposes the query, export and verify-chain subcommands. + /// Verifies that the audit command group exposes the query, export, verify-chain and archive subcommands. [Fact] public void AuditCommandGroup_HasQueryExportAndVerifyChain() { @@ -80,6 +80,7 @@ public class CommandTreeTests Assert.Contains("query", subNames); Assert.Contains("export", subNames); Assert.Contains("verify-chain", subNames); + Assert.Contains("archive", subNames); } /// Verifies that the bundle command group exposes the export, preview and import subcommands. diff --git a/tests/ZB.MOM.WW.ScadaBridge.CLI.Tests/Commands/AuditArchiveCommandTests.cs b/tests/ZB.MOM.WW.ScadaBridge.CLI.Tests/Commands/AuditArchiveCommandTests.cs new file mode 100644 index 00000000..8c0b4651 --- /dev/null +++ b/tests/ZB.MOM.WW.ScadaBridge.CLI.Tests/Commands/AuditArchiveCommandTests.cs @@ -0,0 +1,136 @@ +using System.Net; +using System.Net.Sockets; +using System.Text; +using System.Text.Json; +using ZB.MOM.WW.ScadaBridge.CLI.Commands; + +namespace ZB.MOM.WW.ScadaBridge.CLI.Tests.Commands; + +/// +/// Tests for scadabridge audit archive list / verify (ScadaBridge#42): each +/// sends its management command (ListAuditArchives, VerifyAuditArchive), and +/// verify maps the verdict onto the exit code (0 verified, 3 not verified). +/// +[Collection("Console")] +public class AuditArchiveCommandTests +{ + private const string Name = "auditlog-2025-10-20261109T120000123Z"; + + private const string ListBody = + "{\"enabled\":true,\"archives\":[{\"name\":\"" + Name + "\",\"partition\":\"2025-10\",\"status\":\"Purged\"," + + "\"rowCount\":11,\"chainedRowCount\":10,\"dataBytes\":4096}]}"; + + private const string VerifiedBody = + "{\"name\":\"" + Name + "\",\"partition\":\"2025-10\",\"manifestStatus\":\"Purged\",\"verified\":true,\"filesIntact\":true," + + "\"rowsRead\":11,\"chains\":[{\"chain\":{\"partition\":\"2025-10\",\"status\":\"Intact\",\"rowsVerified\":10," + + "\"tombstonesApplied\":2,\"unchainedRows\":1},\"anchor\":\"Matches\",\"anchorDetail\":\"ends at the anchor\"}]}"; + + private const string MismatchBody = + "{\"name\":\"" + Name + "\",\"partition\":\"2025-10\",\"manifestStatus\":\"Purged\",\"verified\":false,\"filesIntact\":true," + + "\"rowsRead\":11,\"chains\":[{\"chain\":{\"partition\":\"2025-10\",\"status\":\"Intact\",\"rowsVerified\":10," + + "\"tombstonesApplied\":0,\"unchainedRows\":0},\"anchor\":\"Mismatch\",\"anchorDetail\":\"hash differs\"}]}"; + + /// The audit group carries the archive group with list + verify. + [Fact] + public void AuditGroup_HasArchiveListAndVerify() + { + var root = AuditCommandTestHarness.BuildRoot(); + var archive = root.Subcommands.Single(c => c.Name == "audit").Subcommands.Single(c => c.Name == "archive"); + Assert.Equal(["list", "verify"], archive.Subcommands.Select(c => c.Name).Order()); + } + + /// list sends ListAuditArchives and prints the archives. + /// A task that represents the asynchronous operation. + [Fact] + public async Task List_SendsListAuditArchives_AndPrintsTheArchives() + { + var (exit, output, request) = await RunAsync(ListBody, "audit", "archive", "list", "--format", "table"); + + Assert.Equal(0, exit); + Assert.Contains(Name, output); + Assert.Contains("Archiving: enabled", output); + using var payload = JsonDocument.Parse(request); + Assert.Equal("ListAuditArchives", payload.RootElement.GetProperty("command").GetString()); + } + + /// verify sends VerifyAuditArchive with the name and maps the verdict to the exit code. + /// The verification result the fake endpoint serves. + /// Output format. + /// Expected exit code. + /// Text the report must contain. + /// A task that represents the asynchronous operation. + [Theory] + [InlineData(VerifiedBody, "table", 0, "anchor Matches")] + [InlineData(VerifiedBody, "json", 0, "\"verified\":true")] + [InlineData(MismatchBody, "table", 3, "Verified : NO")] + [InlineData(MismatchBody, "json", 3, "\"anchor\":\"Mismatch\"")] + public async Task Verify_SendsVerifyAuditArchive_AndMapsVerdictToExitCode( + string body, string format, int expectedExit, string expectedText) + { + var (exit, output, request) = await RunAsync(body, "audit", "archive", "verify", "--name", Name, "--format", format); + + Assert.Equal(expectedExit, exit); + Assert.Contains(expectedText, output); + using var payload = JsonDocument.Parse(request); + Assert.Equal("VerifyAuditArchive", payload.RootElement.GetProperty("command").GetString()); + var name = payload.RootElement.GetProperty("payload").EnumerateObject() + .First(p => string.Equals(p.Name, "name", StringComparison.OrdinalIgnoreCase)).Value.GetString(); + Assert.Equal(Name, name); + } + + /// verify without --name is a usage error. + [Fact] + public void Verify_MissingName_ProducesRequiredFlagError() + { + var root = AuditCommandTestHarness.BuildRoot(); + var (exit, _, err) = AuditCommandTestHarness.Invoke(root, "audit", "archive", "verify"); + + Assert.NotEqual(0, exit); + Assert.Contains("--name", err); + } + + private static async Task<(int Exit, string Output, string Request)> RunAsync(string body, params string[] args) + { + var listener = new TcpListener(IPAddress.Loopback, 0); + listener.Start(); + try + { + var port = ((IPEndPoint)listener.LocalEndpoint).Port; + string? requestBody = null; + var server = Task.Run(async () => + { + using var client = await listener.AcceptTcpClientAsync(); + await using var stream = client.GetStream(); + using var reader = new StreamReader(stream, Encoding.UTF8, leaveOpen: true); + var length = 0; + while (await reader.ReadLineAsync() is { Length: > 0 } line) + if (line.StartsWith("Content-Length:", StringComparison.OrdinalIgnoreCase)) + length = int.Parse(line[15..].Trim()); + var buffer = new char[length]; + var read = 0; + while (read < length) + { + var count = await reader.ReadAsync(buffer.AsMemory(read)); + if (count == 0) throw new EndOfStreamException(); + read += count; + } + requestBody = new string(buffer); + var bytes = Encoding.UTF8.GetBytes(body); + var headers = Encoding.ASCII.GetBytes( + $"HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: {bytes.Length}\r\nConnection: close\r\n\r\n"); + await stream.WriteAsync(headers); + await stream.WriteAsync(bytes); + }); + + var root = AuditCommandTestHarness.BuildRoot(); + var (exit, output, _) = AuditCommandTestHarness.Invoke( + root, [.. args, "--url", $"http://127.0.0.1:{port}", "--username", "u", "--password", "p"]); + await server.WaitAsync(TimeSpan.FromSeconds(5)); + return (exit, output, requestBody!); + } + finally + { + listener.Stop(); + } + } +} diff --git a/tests/ZB.MOM.WW.ScadaBridge.CLI.Tests/Commands/AuditCommandsScaffoldTests.cs b/tests/ZB.MOM.WW.ScadaBridge.CLI.Tests/Commands/AuditCommandsScaffoldTests.cs index 8b93275e..a2540984 100644 --- a/tests/ZB.MOM.WW.ScadaBridge.CLI.Tests/Commands/AuditCommandsScaffoldTests.cs +++ b/tests/ZB.MOM.WW.ScadaBridge.CLI.Tests/Commands/AuditCommandsScaffoldTests.cs @@ -27,14 +27,14 @@ public class AuditCommandsScaffoldTests Assert.False(string.IsNullOrWhiteSpace(audit.Description)); } - /// Verifies that the audit group exposes the query, export, tree, verify-chain and backfill-source-node subcommands. + /// Verifies that the audit group exposes the query, export, tree, verify-chain, backfill-source-node and archive subcommands. [Fact] - public void Audit_HasFiveSubcommands_QueryExportTreeVerifyChainBackfillSourceNode() + public void Audit_HasSixSubcommands_QueryExportTreeVerifyChainBackfillSourceNodeArchive() { var audit = BuildAudit(); var names = audit.Subcommands.Select(c => c.Name).OrderBy(n => n).ToArray(); Assert.Equal( - new[] { "backfill-source-node", "export", "query", "tree", "verify-chain" }, + new[] { "archive", "backfill-source-node", "export", "query", "tree", "verify-chain" }, names); } @@ -62,8 +62,11 @@ public class AuditCommandsScaffoldTests [Fact] public void Audit_EveryLeafCommand_HasAnAction() { + // `archive` is a group (list / verify, ScadaBridge#42): check its leaves, not it. var audit = BuildAudit(); - Assert.All(audit.Subcommands, sub => + var leaves = audit.Subcommands.SelectMany(c => c.Subcommands.Count == 0 ? [c] : c.Subcommands).ToList(); + Assert.Equal(7, leaves.Count); + Assert.All(leaves, sub => Assert.True(sub.Action != null, $"Leaf command '{sub.Name}' has no action.")); } } diff --git a/tests/ZB.MOM.WW.ScadaBridge.CLI.Tests/ManagementSurfaceParityTests.cs b/tests/ZB.MOM.WW.ScadaBridge.CLI.Tests/ManagementSurfaceParityTests.cs index d65ee09e..0e3d17b4 100644 --- a/tests/ZB.MOM.WW.ScadaBridge.CLI.Tests/ManagementSurfaceParityTests.cs +++ b/tests/ZB.MOM.WW.ScadaBridge.CLI.Tests/ManagementSurfaceParityTests.cs @@ -44,6 +44,9 @@ public class ManagementSurfaceParityTests Assert.Contains("ListSharedScripts", sent.Keys); } + private static readonly HashSet AuditManagementLeaves = + ["audit verify-chain", "audit archive list", "audit archive verify"]; + private static async Task> ProbeProductionTree(string? omitGroup = null) { var root = CliApplication.BuildRoot(); @@ -93,9 +96,10 @@ public class ManagementSurfaceParityTests foreach (var (leaf, path) in Leaves(root, [])) { // Separate REST and persistent SignalR surfaces, not management contracts. - // `audit verify-chain` is the one audit leaf that IS a management contract - // (VerifyAuditChain, ScadaBridge#41), so it stays in the probe. - if (path.Length == 0 || (path[0] == "audit" && string.Join(' ', path) != "audit verify-chain") + // `audit verify-chain` (VerifyAuditChain, ScadaBridge#41) and `audit archive + // list|verify` (ListAuditArchives / VerifyAuditArchive, ScadaBridge#42) are the + // audit leaves that ARE management contracts, so they stay in the probe. + if (path.Length == 0 || (path[0] == "audit" && !AuditManagementLeaves.Contains(string.Join(' ', path))) || path[0] == omitGroup || string.Join(' ', path) == "debug stream") continue; currentPath = string.Join(' ', path); diff --git a/tests/ZB.MOM.WW.ScadaBridge.ManagementService.Tests/AuditArchiveCommandTests.cs b/tests/ZB.MOM.WW.ScadaBridge.ManagementService.Tests/AuditArchiveCommandTests.cs new file mode 100644 index 00000000..d69a04ec --- /dev/null +++ b/tests/ZB.MOM.WW.ScadaBridge.ManagementService.Tests/AuditArchiveCommandTests.cs @@ -0,0 +1,147 @@ +using System.Text.Json; +using Akka.Actor; +using Akka.TestKit.Xunit2; +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Logging.Abstractions; +using NSubstitute; +using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Repositories; +using ZB.MOM.WW.ScadaBridge.Commons.Messages.Management; +using ZB.MOM.WW.ScadaBridge.Commons.Types.Audit; + +namespace ZB.MOM.WW.ScadaBridge.ManagementService.Tests; + +/// +/// The ListAuditArchives / VerifyAuditArchive management commands +/// (ScadaBridge#42): role gate, site-scope refusal, an unconfigured archive directory, bad +/// names, and the results reaching the wire with enums as names. +/// +public class AuditArchiveCommandTests : TestKit +{ + private const string Name = "auditlog-2025-10-20261109T120000123Z"; + + private readonly IAuditLogArchive _archive = Substitute.For(); + + /// Configures the archive substitute as a configured, enabled archive. + public AuditArchiveCommandTests() + { + _archive.IsConfigured.Returns(true); + _archive.IsEnabled.Returns(true); + } + + private IActorRef CreateActor(bool registerArchive = true) + { + var services = new ServiceCollection().AddTestConfigurationTransaction(); + if (registerArchive) + { + services.AddScoped(_ => _archive); + } + + return Sys.ActorOf(Props.Create(() => new ManagementActor( + services.BuildServiceProvider(), NullLogger.Instance))); + } + + private static ManagementEnvelope Envelope(object command, string[] sites, params string[] roles) => + new(new AuthenticatedUser("u", "U", roles, sites), command, Guid.NewGuid().ToString("N")); + + private static AuditArchiveManifest Manifest() => new( + 1, Name, new DateTime(2025, 10, 1, 0, 0, 0, DateTimeKind.Utc), "2025-10", + new DateTime(2026, 11, 9, 12, 0, 0, DateTimeKind.Utc), AuditArchiveStatus.Purged, + new DateTime(2026, 11, 9, 12, 0, 1, DateTimeKind.Utc), 10, 9, 1, null, null, + Name + ".parquet", "ab", 1234, Name + ".tombstones.parquet", "cd", 0, + [new AuditArchiveChainState("2025-10", 9, new string('0', 64), null, null, 0)]); + + /// A Viewer lists the archives; status renders as a name. + [Fact] + public void Viewer_ListsArchives() + { + _archive.ListAsync(Arg.Any()).Returns([Manifest()]); + + CreateActor().Tell(Envelope(new ListAuditArchivesCommand(), [], "Viewer")); + + var response = ExpectMsg(TimeSpan.FromSeconds(5)); + using var json = JsonDocument.Parse(response.JsonData); + Assert.True(json.RootElement.GetProperty("enabled").GetBoolean()); + var archive = json.RootElement.GetProperty("archives")[0]; + Assert.Equal(Name, archive.GetProperty("name").GetString()); + Assert.Equal("Purged", archive.GetProperty("status").GetString()); + } + + /// Verification results reach the wire with the anchor and chain status as names. + [Fact] + public void Viewer_GetsVerificationResult() + { + var chain = new AuditChainVerificationResult("2025-10", AuditChainStatus.Intact, 9, 0, null, 9, 1, null); + _archive.VerifyAsync(Name, Arg.Any()).Returns(new AuditArchiveVerificationResult( + Name, "2025-10", AuditArchiveStatus.Purged, true, true, null, 10, + [new AuditArchiveChainVerification(chain, AuditArchiveAnchorStatus.Matches, "ends at the anchor")])); + + CreateActor().Tell(Envelope(new VerifyAuditArchiveCommand(Name), [], "Viewer")); + + var response = ExpectMsg(TimeSpan.FromSeconds(5)); + using var json = JsonDocument.Parse(response.JsonData); + Assert.True(json.RootElement.GetProperty("verified").GetBoolean()); + var c = json.RootElement.GetProperty("chains")[0]; + Assert.Equal("Matches", c.GetProperty("anchor").GetString()); + Assert.Equal("Intact", c.GetProperty("chain").GetProperty("status").GetString()); + } + + /// Designer/Deployer without an OperationalAudit role are refused both commands. + [Fact] + public void NonAuditRole_IsUnauthorized() + { + var actor = CreateActor(); + actor.Tell(Envelope(new ListAuditArchivesCommand(), [], "Designer", "Deployer")); + ExpectMsg(TimeSpan.FromSeconds(5)); + actor.Tell(Envelope(new VerifyAuditArchiveCommand(Name), [], "Designer", "Deployer")); + ExpectMsg(TimeSpan.FromSeconds(5)); + _archive.DidNotReceiveWithAnyArgs().ListAsync(default); + _archive.DidNotReceiveWithAnyArgs().VerifyAsync(default!, default); + } + + /// A site-scoped Viewer is refused: an archive holds every site's rows. + [Fact] + public void SiteScopedViewer_IsRefused() + { + CreateActor().Tell(Envelope(new VerifyAuditArchiveCommand(Name), ["3"], "Viewer")); + + var response = ExpectMsg(TimeSpan.FromSeconds(5)); + Assert.Contains("every site", response.Message); + _archive.DidNotReceiveWithAnyArgs().VerifyAsync(default!, default); + } + + /// No archive directory configured is an explicit error code, not an empty list. + [Fact] + public void UnconfiguredArchive_IsArchiveNotConfigured() + { + _archive.IsConfigured.Returns(false); + + CreateActor().Tell(Envelope(new ListAuditArchivesCommand(), [], "Administrator")); + + var response = ExpectMsg(TimeSpan.FromSeconds(5)); + Assert.Equal("ARCHIVE_NOT_CONFIGURED", response.ErrorCode); + } + + /// A malformed name is INVALID_ARGUMENT; an unknown one is NOT_FOUND. + [Fact] + public void BadNames_MapToCallerErrors() + { + _archive.VerifyAsync("../etc", Arg.Any()) + .Returns>(_ => throw new ArgumentException("not an archive name")); + _archive.VerifyAsync(Name, Arg.Any()) + .Returns>(_ => throw new FileNotFoundException("no archive")); + + var actor = CreateActor(); + actor.Tell(Envelope(new VerifyAuditArchiveCommand("../etc"), [], "Administrator")); + Assert.Equal("INVALID_ARGUMENT", ExpectMsg(TimeSpan.FromSeconds(5)).ErrorCode); + actor.Tell(Envelope(new VerifyAuditArchiveCommand(Name), [], "Administrator")); + Assert.Equal("NOT_FOUND", ExpectMsg(TimeSpan.FromSeconds(5)).ErrorCode); + } + + /// Archive verification runs under the long-running Ask timeout. + [Fact] + public void VerifyAuditArchive_UsesLongRunningAskTimeout() + { + Assert.Equal(TimeSpan.FromMinutes(5), + ManagementEndpoints.ResolveAskTimeout(new ManagementServiceOptions(), typeof(VerifyAuditArchiveCommand))); + } +} diff --git a/tests/ZB.MOM.WW.ScadaBridge.ManagementService.Tests/AuditEndpointsTests.cs b/tests/ZB.MOM.WW.ScadaBridge.ManagementService.Tests/AuditEndpointsTests.cs index 4992b5d3..2155e718 100644 --- a/tests/ZB.MOM.WW.ScadaBridge.ManagementService.Tests/AuditEndpointsTests.cs +++ b/tests/ZB.MOM.WW.ScadaBridge.ManagementService.Tests/AuditEndpointsTests.cs @@ -506,21 +506,43 @@ public class AuditEndpointsTests Assert.Equal(propertyNames, AuditEndpoints.CsvHeader.Split(',')); } - /// Verifies that a Parquet export returns 501. + /// + /// format=parquet (ScadaBridge#42; it used to be a 501) streams one readable Parquet + /// file whose columns are the csv/jsonl export's columns, in order, holding every row. + /// /// A task that represents the asynchronous operation. [Fact] - public async Task Export_Parquet_Returns501() + public async Task Export_Parquet_StreamsReadableParquet_WithTheCsvColumns() { - // Parquet archival is deferred to v1.x (Component-AuditLog.md) — no - // library is referenced, so the endpoint returns 501 with guidance. - var (client, _, host) = await BuildHostAsync(roles: new[] { "Administrator" }); + var (client, _, host) = await BuildHostAsync( + roles: new[] { "Administrator" }, + queryPages: new[] + { + (IReadOnlyList)new[] + { + SampleEvent(Guid.Parse("cccccccc-0000-0000-0000-000000000001")), + SampleEvent(Guid.Parse("cccccccc-0000-0000-0000-000000000002")), + }, + (IReadOnlyList)Array.Empty(), + }); using (host) { - var response = await client.SendAsync(Get("/api/audit/export?format=parquet")); + var response = await client.SendAsync(Get("/api/audit/export?format=Parquet")); - Assert.Equal(HttpStatusCode.NotImplemented, response.StatusCode); - var body = await response.Content.ReadAsStringAsync(); - Assert.Contains("Parquet export deferred to v1.x", body); + Assert.Equal(HttpStatusCode.OK, response.StatusCode); + Assert.Equal(AuditEndpoints.ParquetContentType, response.Content.Headers.ContentType!.MediaType); + Assert.Contains(".parquet", response.Content.Headers.ContentDisposition?.ToString() + ?? response.Headers.GetValues("Content-Disposition").FirstOrDefault() + ?? response.Content.Headers.GetValues("Content-Disposition").First()); + + var bytes = await response.Content.ReadAsByteArrayAsync(); + await using var reader = await Parquet.ParquetReader.CreateAsync(new MemoryStream(bytes)); + Assert.Equal(AuditEndpoints.CsvHeader.Split(','), reader.Schema.GetDataFields().Select(f => f.Name)); + using var group = reader.OpenRowGroupReader(0); + Assert.Equal(2, group.RowCount); + var ids = new string?[2]; + await group.ReadAsync(reader.Schema.GetDataFields()[0], ids!); + Assert.Equal(new[] { "cccccccc-0000-0000-0000-000000000001", "cccccccc-0000-0000-0000-000000000002" }, ids); } } diff --git a/tests/ZB.MOM.WW.ScadaBridge.ManagementService.Tests/AuditExportParquetTests.cs b/tests/ZB.MOM.WW.ScadaBridge.ManagementService.Tests/AuditExportParquetTests.cs new file mode 100644 index 00000000..3c3fe68e --- /dev/null +++ b/tests/ZB.MOM.WW.ScadaBridge.ManagementService.Tests/AuditExportParquetTests.cs @@ -0,0 +1,150 @@ +using Parquet; +using ZB.MOM.WW.ScadaBridge.Commons.Types.Audit; +using ZB.MOM.WW.ScadaBridge.Commons.Types.Enums; + +namespace ZB.MOM.WW.ScadaBridge.ManagementService.Tests; + +/// +/// (ScadaBridge#42): the Parquet export round-trips every +/// one of the 24 export columns, splits into row groups, and never writes synchronously to +/// the output (Kestrel rejects synchronous response writes). +/// +public class AuditExportParquetTests +{ + /// Rows written and read back are equal, column for column, across row groups. + /// A task that represents the asynchronous operation. + [Fact] + public async Task RoundTrip_EveryColumn_AcrossRowGroups() + { + var rows = Enumerable.Range(0, AuditExportParquet.RowGroupRows + 7).Select(Row).ToList(); + var output = new AsyncOnlyStream(); + + var written = await AuditExportParquet.WriteAsync(Pages(rows, 1000), output); + + Assert.Equal(rows.Count, written); + await using var reader = await ParquetReader.CreateAsync(new MemoryStream(output.ToArray())); + Assert.Equal(2, reader.RowGroupCount); + var back = new List(); + var fields = reader.Schema.GetDataFields(); + for (var g = 0; g < reader.RowGroupCount; g++) + { + using var group = reader.OpenRowGroupReader(g); + var n = (int)group.RowCount; + async Task S(int i) { var a = new string?[n]; await group.ReadAsync(fields[i], a!); return a; } + async Task V(int i) where T : struct { var a = new T?[n]; await group.ReadAsync(fields[i], a); return a; } + var id = await S(0); var occurred = await V(1); var ingested = await V(2); + var channel = await S(3); var kind = await S(4); var corr = await S(5); var exec = await S(6); var parent = await S(7); + var site = await S(8); var node = await S(9); var inst = await S(10); var script = await S(11); var actor = await S(12); + var target = await S(13); var status = await S(14); var http = await V(15); var dur = await V(16); + var errMsg = await S(17); var errDetail = await S(18); var req = await S(19); var resp = await S(20); + var trunc = await V(21); var extra = await S(22); var fwd = await S(23); + for (var i = 0; i < n; i++) + { + back.Add(new AuditExportRow + { + EventId = Guid.Parse(id[i]!), + OccurredAtUtc = occurred[i]!.Value, + IngestedAtUtc = ingested[i], + Channel = Enum.Parse(channel[i]!), + Kind = Enum.Parse(kind[i]!), + CorrelationId = corr[i] is { } c ? Guid.Parse(c) : null, + ExecutionId = exec[i] is { } e ? Guid.Parse(e) : null, + ParentExecutionId = parent[i] is { } p ? Guid.Parse(p) : null, + SourceSiteId = site[i], + SourceNode = node[i], + SourceInstanceId = inst[i], + SourceScript = script[i], + Actor = actor[i], + Target = target[i], + Status = Enum.Parse(status[i]!), + HttpStatus = http[i], + DurationMs = dur[i], + ErrorMessage = errMsg[i], + ErrorDetail = errDetail[i], + RequestSummary = req[i], + ResponseSummary = resp[i], + PayloadTruncated = trunc[i]!.Value, + Extra = extra[i], + ForwardState = fwd[i] is { } f ? Enum.Parse(f) : null, + }); + } + } + + Assert.Equal(rows, back); + } + + /// An export with no rows is still a valid (empty) Parquet file. + /// A task that represents the asynchronous operation. + [Fact] + public async Task NoRows_IsAValidEmptyFile() + { + var output = new AsyncOnlyStream(); + Assert.Equal(0, await AuditExportParquet.WriteAsync(Pages([], 10), output)); + await using var reader = await ParquetReader.CreateAsync(new MemoryStream(output.ToArray())); + Assert.Equal(0, reader.RowGroupCount); + Assert.Equal(24, reader.Schema.GetDataFields().Length); + } + + private static AuditExportRow Row(int i) => new() + { + EventId = Guid.NewGuid(), + // Microsecond precision: the export timestamp column is TIMESTAMP(MICROS). + OccurredAtUtc = new DateTime(2026, 5, 20, 12, 0, 0, DateTimeKind.Utc).AddTicks(i * 10L), + IngestedAtUtc = i % 3 == 0 ? null : new DateTime(2026, 5, 20, 12, 0, 1, DateTimeKind.Utc), + Channel = (AuditChannel)(i % Enum.GetValues().Length), + Kind = AuditKind.ApiCall, + CorrelationId = i % 2 == 0 ? Guid.NewGuid() : null, + ExecutionId = Guid.NewGuid(), + ParentExecutionId = i % 5 == 0 ? Guid.NewGuid() : null, + SourceSiteId = i % 4 == 0 ? null : "plant-a", + SourceNode = "node-a", + SourceInstanceId = "Plant1.Boiler", + SourceScript = i % 2 == 0 ? "OnTick" : null, + Actor = i % 7 == 0 ? string.Empty : "api-key-1", + Target = "erp", + Status = AuditStatus.Delivered, + HttpStatus = i % 2 == 0 ? 200 : null, + DurationMs = i, + ErrorMessage = i % 9 == 0 ? "boom, \"quoted\"\nline" : null, + ErrorDetail = null, + RequestSummary = "{\"ü\":\"日本\"}", + ResponseSummary = null, + PayloadTruncated = i % 2 == 1, + Extra = null, + ForwardState = i % 11 == 0 ? AuditForwardState.Forwarded : null, + }; + + private static async IAsyncEnumerable> Pages(List rows, int size) + { + for (var i = 0; i < rows.Count; i += size) + { + await Task.Yield(); + yield return rows.Skip(i).Take(size).ToList(); + } + } + + /// A response-body stand-in that fails any synchronous write, like Kestrel. + private sealed class AsyncOnlyStream : Stream + { + private readonly MemoryStream _inner = new(); + + public byte[] ToArray() => _inner.ToArray(); + + public override bool CanRead => false; + public override bool CanSeek => false; + public override bool CanWrite => true; + public override long Length => throw new NotSupportedException(); + public override long Position { get => throw new NotSupportedException(); set => throw new NotSupportedException(); } + public override void Flush() => throw new InvalidOperationException("Synchronous operations are disallowed."); + public override Task FlushAsync(CancellationToken cancellationToken) => Task.CompletedTask; + public override int Read(byte[] buffer, int offset, int count) => throw new NotSupportedException(); + public override long Seek(long offset, SeekOrigin origin) => throw new NotSupportedException(); + public override void SetLength(long value) => throw new NotSupportedException(); + public override void Write(byte[] buffer, int offset, int count) => throw new InvalidOperationException("Synchronous operations are disallowed."); + public override ValueTask WriteAsync(ReadOnlyMemory buffer, CancellationToken cancellationToken = default) + { + _inner.Write(buffer.Span); + return ValueTask.CompletedTask; + } + } +} diff --git a/tests/ZB.MOM.WW.ScadaBridge.ManagementService.Tests/RequiredRoleMatrixTests.cs b/tests/ZB.MOM.WW.ScadaBridge.ManagementService.Tests/RequiredRoleMatrixTests.cs index 1f44a349..afd85b6a 100644 --- a/tests/ZB.MOM.WW.ScadaBridge.ManagementService.Tests/RequiredRoleMatrixTests.cs +++ b/tests/ZB.MOM.WW.ScadaBridge.ManagementService.Tests/RequiredRoleMatrixTests.cs @@ -171,6 +171,9 @@ public class RequiredRoleMatrixTests // AuditLog hash-chain verification (ScadaBridge#41) carries audit query's // OperationalAudit permission (alog.md §15.1); written out by hand to keep the freeze. ["VerifyAuditChain"] = [Roles.Administrator, Roles.Viewer], + // AuditLog Parquet archives (ScadaBridge#42): same audit-read gate as verify-chain. + ["ListAuditArchives"] = [Roles.Administrator, Roles.Viewer], + ["VerifyAuditArchive"] = [Roles.Administrator, Roles.Viewer], // ---- Two-person secured write ---------------------------------------------- ["SubmitSecuredWrite"] = [Roles.Operator],