dohertj2/ScadaBridge
1
0
Fork 0
Code Issues 2 Pull Requests Actions Packages Projects Releases Wiki Activity

feat(inbound-api): audit central method-script Database.* calls (DbOutbound) and feed the flow-trace sink

Browse Source 
InboundDatabaseHelper gains trailing optional auditWriter / nodeIdentity /
executionId / parentExecutionId / methodName / traceSink parameters; every
QuerySingle/Query/Execute writes one DbOutbound/DbWrite row (Delivered/Failed,
Target = connection name, Actor = 'api:{method}', DurationMs, SQL + parameters
as RequestSummary, Extra in the site's {op, rowsAffected|rowsReturned}
vocabulary, SourceNode) via ICentralAuditWriter — best-effort, never into the
script; the same instance goes to the RouteHelper trace sink first. Closes the
audit-scope gap Component-AuditLog.md already claimed (plan D7). Redaction and
the 8 KB cap apply at write time in the shared redactor, as for site rows.

Claude-Session: https://claude.ai/code/session_01NcMG6xXdsJbmD2wYAgkK29
This commit is contained in:
Joseph Doherty
2026-09-06 01:52:50 -04:00
parent 3b9def5a09
commit 69feb40e65
4 changed files with 514 additions and 25 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/ZB.MOM.WW.ScadaBridge.InboundAPI/InboundDatabaseHelper.cs
+249 -24
View File
@@ -1,6 +1,11 @@
using System.Data.Common;
using System.Diagnostics;
using System.Globalization;
using System.Text;
using ZB.MOM.WW.Audit;
using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Services;
using ZB.MOM.WW.ScadaBridge.Commons.Types.Audit;
using ZB.MOM.WW.ScadaBridge.Commons.Types.Enums;
namespace ZB.MOM.WW.ScadaBridge.InboundAPI;
@@ -36,12 +41,35 @@ namespace ZB.MOM.WW.ScadaBridge.InboundAPI;
/// backstop derived from the method timeout — so a slow query is bounded by the method
/// timeout instead of running unbounded.
/// </para>
///
/// <para>
/// <b>Audited (Audit Log #23, outbound DB scope; Flow Debugger outbound-calls plan
/// 2026-09-06 D7).</b> Every call emits exactly ONE <c>DbOutbound</c>/<c>DbWrite</c> row
/// through the central <see cref="ICentralAuditWriter"/> using the site's
/// <c>AuditingDbCommand</c> vocabulary: <c>Extra = {"op":"write","rowsAffected":N}</c> for
/// <see cref="ExecuteAsync"/>, <c>{"op":"read","rowsReturned":N}</c> for the two query
/// methods; <c>RequestSummary</c> = <c>{"sql":…,"parameters":{"@name":"value"}}</c>. The
/// 8 KB payload cap and per-target SQL-parameter redaction are applied at write time by
/// the writer's <c>IAuditRedactor</c> — the same stage that caps the site's rows — so the
/// helper builds the raw summary only. Emission is best-effort: a build, write or sink
/// fault never reaches the script; the original result (or exception) always does. When
/// a flow-trace sink is present the same <see cref="AuditEvent"/> instance goes to the
/// sink first, then to the writer, so a Live Flow Debugger run renders the row without
/// a round-trip to the audit store. A helper built without a writer and without a sink
/// (the pre-existing three-argument shape) emits nothing.
/// </para>
/// </summary>
public sealed class InboundDatabaseHelper
{
private readonly IDatabaseGateway? _gateway;
private readonly CancellationToken _ct;
private readonly int _commandTimeoutSeconds;
private readonly ICentralAuditWriter? _auditWriter;
private readonly INodeIdentityProvider? _nodeIdentity;
private readonly Guid? _executionId;
private readonly Guid? _parentExecutionId;
private readonly string? _actor;
private readonly Action<AuditEvent>? _traceSink;
/// <summary>
/// Initializes the helper.
@@ -49,13 +77,36 @@ public sealed class InboundDatabaseHelper
/// <param name="gateway">The central database gateway, or null when no gateway is registered (the helper then throws on first use).</param>
/// <param name="ct">The executing method's deadline token; forwarded to every async DB call so a slow query is bounded by the method timeout.</param>
/// <param name="commandTimeout">The method timeout; used to derive a <see cref="DbCommand.CommandTimeout"/> backstop. <see cref="TimeSpan.Zero"/> (the default) leaves the provider default.</param>
public InboundDatabaseHelper(IDatabaseGateway? gateway, CancellationToken ct, TimeSpan commandTimeout = default)
/// <param name="auditWriter">Central-direct-write audit writer for the per-call <c>DbOutbound</c>/<c>DbWrite</c> row; null = no row is written.</param>
/// <param name="nodeIdentity">Stamps <c>SourceNode</c> (central-a/central-b) on the row; null = NULL.</param>
/// <param name="executionId">The executing inbound request's per-request execution id, stamped as the row's <c>ExecutionId</c>.</param>
/// <param name="parentExecutionId">The spawning execution's id, if any (an inbound request is a root, so callers normally pass null).</param>
/// <param name="methodName">The executing API method's name; the row's <c>Actor</c> becomes <c>api:{methodName}</c>. Null = NULL actor.</param>
/// <param name="traceSink">Flow Debugger trace sink; receives the same event instance before the writer does. Null = none.</param>
// Trailing optional parameters: the original (gateway, ct, commandTimeout) shape stays
// source-compatible for every existing caller and test.
public InboundDatabaseHelper(
IDatabaseGateway? gateway,
CancellationToken ct,
TimeSpan commandTimeout = default,
ICentralAuditWriter? auditWriter = null,
INodeIdentityProvider? nodeIdentity = null,
Guid? executionId = null,
Guid? parentExecutionId = null,
string? methodName = null,
Action<AuditEvent>? traceSink = null)
{
_gateway = gateway;
_ct = ct;
_commandTimeoutSeconds = commandTimeout > TimeSpan.Zero
? (int)Math.Ceiling(commandTimeout.TotalSeconds)
: 0;
_auditWriter = auditWriter;
_nodeIdentity = nodeIdentity;
_executionId = executionId;
_parentExecutionId = parentExecutionId;
_actor = string.IsNullOrEmpty(methodName) ? null : "api:" + methodName;
_traceSink = traceSink;
}
/// <summary>First column of the first row converted to <typeparamref name="T"/> (default if no rows).</summary>
@@ -66,13 +117,16 @@ public sealed class InboundDatabaseHelper
/// <returns>The converted scalar, or <c>default</c> when there are no rows / a NULL value.</returns>
public async Task<T?> QuerySingleAsync<T>(string connectionName, string sql, object? parameters = null)
{
if (_gateway is null) throw new InvalidOperationException("Database is not available for this inbound method");
await using var conn = await _gateway.GetConnectionAsync(connectionName, _ct);
await using var cmd = CreateCommand(conn, sql, parameters);
var result = await cmd.ExecuteScalarAsync(_ct);
if (result is null or DBNull) return default;
if (result is T t) return t;
return (T)Convert.ChangeType(result, typeof(T), CultureInfo.InvariantCulture);
var (result, _) = await RunAudited(connectionName, sql, parameters, op: "read", async () =>
{
await using var conn = await _gateway!.GetConnectionAsync(connectionName, _ct);
await using var cmd = CreateCommand(conn, sql, parameters);
var scalar = await cmd.ExecuteScalarAsync(_ct);
if (scalar is null or DBNull) return (default(T), 0);
if (scalar is T t) return (t, 1);
return ((T)Convert.ChangeType(scalar, typeof(T), CultureInfo.InvariantCulture), 1);
});
return result;
}
/// <summary>All rows as column→value dictionaries (case-insensitive keys).</summary>
@@ -83,21 +137,24 @@ public sealed class InboundDatabaseHelper
public async Task<IReadOnlyList<IReadOnlyDictionary<string, object?>>> QueryAsync(
string connectionName, string sql, object? parameters = null)
{
if (_gateway is null) throw new InvalidOperationException("Database is not available for this inbound method");
await using var conn = await _gateway.GetConnectionAsync(connectionName, _ct);
await using var cmd = CreateCommand(conn, sql, parameters);
await using var reader = await cmd.ExecuteReaderAsync(_ct);
var rows = new List<IReadOnlyDictionary<string, object?>>();
while (await reader.ReadAsync(_ct))
var (rows, _) = await RunAudited(connectionName, sql, parameters, op: "read", async () =>
{
var row = new Dictionary<string, object?>(StringComparer.OrdinalIgnoreCase);
for (var i = 0; i < reader.FieldCount; i++)
await using var conn = await _gateway!.GetConnectionAsync(connectionName, _ct);
await using var cmd = CreateCommand(conn, sql, parameters);
await using var reader = await cmd.ExecuteReaderAsync(_ct);
var list = new List<IReadOnlyDictionary<string, object?>>();
while (await reader.ReadAsync(_ct))
{
var v = reader.GetValue(i);
row[reader.GetName(i)] = v is DBNull ? null : v;
var row = new Dictionary<string, object?>(StringComparer.OrdinalIgnoreCase);
for (var i = 0; i < reader.FieldCount; i++)
{
var v = reader.GetValue(i);
row[reader.GetName(i)] = v is DBNull ? null : v;
}
list.Add(row);
}
rows.Add(row);
}
return ((IReadOnlyList<IReadOnlyDictionary<string, object?>>)list, list.Count);
});
return rows;
}
@@ -112,10 +169,178 @@ public sealed class InboundDatabaseHelper
/// <returns>The number of rows affected.</returns>
public async Task<int> ExecuteAsync(string connectionName, string sql, object? parameters = null)
{
if (_gateway is null) throw new InvalidOperationException("Database is not available for this inbound method");
await using var conn = await _gateway.GetConnectionAsync(connectionName, _ct);
await using var cmd = CreateCommand(conn, sql, parameters);
return await cmd.ExecuteNonQueryAsync(_ct);
var (affected, _) = await RunAudited(connectionName, sql, parameters, op: "write", async () =>
{
await using var conn = await _gateway!.GetConnectionAsync(connectionName, _ct);
await using var cmd = CreateCommand(conn, sql, parameters);
var n = await cmd.ExecuteNonQueryAsync(_ct);
return (n, n);
});
return affected;
}
/// <summary>
/// Runs <paramref name="body"/> and emits the one <c>DbOutbound</c>/<c>DbWrite</c> row for
/// it — <c>Delivered</c> with the body's row count, or <c>Failed</c> with the thrown
/// exception's message, which is then rethrown untouched to the script. The gateway
/// null-check is inside the audited span deliberately so "Database is not available"
/// is itself recorded as a failed outbound DB call.
/// </summary>
private async Task<(T Result, int Rows)> RunAudited<T>(
string connectionName, string sql, object? parameters, string op,
Func<Task<(T Result, int Rows)>> body)
{
var startedUtc = DateTime.UtcNow;
var sw = Stopwatch.StartNew();
(T Result, int Rows) outcome;
try
{
if (_gateway is null) throw new InvalidOperationException("Database is not available for this inbound method");
outcome = await body();
}
catch (Exception ex)
{
EmitAudit(connectionName, sql, parameters, op, startedUtc, sw.ElapsedMilliseconds, rows: 0, thrown: ex);
throw;
}
EmitAudit(connectionName, sql, parameters, op, startedUtc, sw.ElapsedMilliseconds, outcome.Rows, thrown: null);
return outcome;
}
/// <summary>
/// Best-effort emission (mirrors <c>RouteTarget.EmitRouteCall</c> and the site's
/// <c>AuditingDbCommand.EmitAudit</c>): build, sink, write — every layer swallows its
/// own faults so nothing reaches the script.
/// </summary>
private void EmitAudit(
string connectionName, string sql, object? parameters, string op,
DateTime startedUtc, long elapsedMs, int rows, Exception? thrown)
{
if (_auditWriter is null && _traceSink is null) return;
AuditEvent evt;
try
{
// Vocabulary lock with the site's AuditingDbCommand: both reads and writes are
// DbWrite rows on the DbOutbound channel; Extra carries the discriminator.
var extra = op == "write"
? $"{{\"op\":\"write\",\"rowsAffected\":{rows}}}"
: $"{{\"op\":\"read\",\"rowsReturned\":{rows}}}";
evt = ScadaBridgeAuditEventFactory.Create(
channel: AuditChannel.DbOutbound,
kind: AuditKind.DbWrite,
status: thrown is null ? AuditStatus.Delivered : AuditStatus.Failed,
occurredAtUtc: DateTime.SpecifyKind(startedUtc, DateTimeKind.Utc),
actor: _actor,
target: connectionName,
sourceNode: _nodeIdentity?.NodeName,
// A sync one-shot DB call has no operation lifecycle, so CorrelationId stays
// null; ExecutionId ties the row to the inbound request's other rows.
correlationId: null,
executionId: _executionId,
parentExecutionId: _parentExecutionId,
durationMs: (int)Math.Min(elapsedMs, int.MaxValue),
errorMessage: thrown?.Message,
errorDetail: thrown?.ToString(),
requestSummary: BuildRequestSummary(sql, parameters),
responseSummary: null,
payloadTruncated: false,
extra: extra);
}
catch (Exception)
{
return;
}
// Sink FIRST (task contract): a Live Flow Debugger run reads the row from the
// sink, and the writer's redaction stage must not be the thing gating it.
if (_traceSink is not null)
{
try { _traceSink(evt); }
catch (Exception) { /* a misbehaving sink must not fail the script */ }
}
if (_auditWriter is null) return;
try
{
var t = _auditWriter.WriteAsync(evt, CancellationToken.None);
if (!t.IsCompletedSuccessfully)
{
_ = t.ContinueWith(static _ => { }, TaskContinuationOptions.OnlyOnFaulted);
}
}
catch (Exception)
{
// Best-effort: an audit failure never fails the DB call.
}
}
/// <summary>
/// Same JSON shape as the site's <c>AuditingDbCommand.BuildRequestSummary</c>:
/// <c>{"sql":"…","parameters":{"@name":"value",…}}</c>, values stringified, NULL as
/// JSON null. Built from the script's arguments rather than a <see cref="DbCommand"/>
/// so a failure before the command exists (gateway/open) still carries the statement.
/// Capping and per-target parameter redaction happen at write time in the redactor.
/// </summary>
private static string? BuildRequestSummary(string sql, object? parameters)
{
if (string.IsNullOrEmpty(sql)) return null;
var sb = new StringBuilder(sql.Length + 64);
sb.Append("{\"sql\":");
AppendJsonString(sb, sql);
var props = parameters?.GetType().GetProperties();
if (props is { Length: > 0 })
{
sb.Append(",\"parameters\":{");
var first = true;
foreach (var prop in props)
{
if (!first) sb.Append(',');
first = false;
AppendJsonString(sb, "@" + prop.Name);
sb.Append(':');
var value = prop.GetValue(parameters);
if (value is null || value is DBNull)
{
sb.Append("null");
}
else
{
AppendJsonString(sb, Convert.ToString(value, CultureInfo.InvariantCulture) ?? string.Empty);
}
}
sb.Append('}');
}
sb.Append('}');
return sb.ToString();
}
private static void AppendJsonString(StringBuilder sb, string value)
{
sb.Append('"');
foreach (var ch in value)
{
switch (ch)
{
case '"': sb.Append("\\\""); break;
case '\\': sb.Append("\\\\"); break;
case '\b': sb.Append("\\b"); break;
case '\f': sb.Append("\\f"); break;
case '\n': sb.Append("\\n"); break;
case '\r': sb.Append("\\r"); break;
case '\t': sb.Append("\\t"); break;
default:
if (ch < 0x20) sb.Append("\\u").Append(((int)ch).ToString("x4"));
else sb.Append(ch);
break;
}
}
sb.Append('"');
}
private DbCommand CreateCommand(DbConnection conn, string sql, object? parameters)
src/ZB.MOM.WW.ScadaBridge.InboundAPI/InboundScriptExecutor.cs
+17 -1
View File
@@ -381,7 +381,23 @@ public class InboundScriptExecutor
// Pass the method timeout so the helper derives a
// CommandTimeout backstop and forwards cts.Token to every async DB call —
// a slow query is then bounded by the method deadline.
var dbHelper = new InboundDatabaseHelper(gateway, cts.Token, timeout);
//
// Flow Debugger outbound calls (plan 2026-09-06, D7): the helper audits every
// Database.* call as a DbOutbound/DbWrite row. The inbound request's own
// execution id (the parentExecutionId argument — see its doc above) is the
// row's ExecutionId, exactly as the RouteCall row stamps it; an inbound request
// is a root, so the row's ParentExecutionId is null. The writer + node identity
// come from the same scope as the gateway (both null-tolerant: an unregistered
// writer means no row, never a failed script), and the trace sink is the one the
// caller attached to the RouteHelper so a Live run collects both row kinds.
var dbHelper = new InboundDatabaseHelper(
gateway, cts.Token, timeout,
auditWriter: scope?.ServiceProvider.GetService<ICentralAuditWriter>(),
nodeIdentity: scope?.ServiceProvider.GetService<INodeIdentityProvider>(),
executionId: parentExecutionId,
parentExecutionId: null,
methodName: method.Name,
traceSink: route.TraceSink);
// Bind the route helper to the method deadline so a
// routed Route.To(...).Call(...) inherits the method-level timeout
src/ZB.MOM.WW.ScadaBridge.InboundAPI/RouteHelper.cs
+8
View File
@@ -164,6 +164,14 @@ public class RouteHelper
new(_instanceLocator, _instanceRouter, _deadlineToken, _requestAbortedToken, _parentExecutionId,
_auditWriter, _nodeIdentity, _trace, traceSink);
/// <summary>
/// Flow Debugger outbound calls (plan 2026-09-06, D5/D7): the trace sink this helper carries,
/// so <see cref="InboundScriptExecutor"/> can hand the SAME delegate to the script's
/// <see cref="InboundDatabaseHelper"/> and a Live run's central <c>Database.*</c> rows land
/// in the same collector as its <c>RouteCall</c> rows. Null when no sink was attached.
/// </summary>
public Action<AuditEvent>? TraceSink => _traceSink;
/// <summary>
/// Creates a route target for the specified instance.
/// </summary>