Joseph Doherty
f5479f3ca3
Add a nullable string? OwnerKeyId property to GatewaySession that captures the API key identifier (KeyId) of the authenticated caller that opened the session. Wire it through ISessionManager.OpenSessionAsync → SessionManager → GatewaySession constructor. The gRPC service passes identityAccessor .Current?.KeyId; internal callers (GatewayAlarmMonitor, DashboardLiveDataService) pass null. Covers the positive and null cases with two new TDD-first tests.
976 lines
38 KiB
C#
976 lines
38 KiB
C#
using System.Diagnostics;
|
|
using Grpc.Core;
|
|
using Google.Protobuf.WellKnownTypes;
|
|
using ZB.MOM.WW.MxGateway.Contracts;
|
|
using ZB.MOM.WW.MxGateway.Contracts.Proto;
|
|
using ZB.MOM.WW.MxGateway.Server.Alarms;
|
|
using ZB.MOM.WW.MxGateway.Server.Metrics;
|
|
using ZB.MOM.WW.MxGateway.Server.Security.Authentication;
|
|
using ZB.MOM.WW.MxGateway.Server.Security.Authorization;
|
|
using ZB.MOM.WW.MxGateway.Server.Sessions;
|
|
using ZB.MOM.WW.MxGateway.Server.Workers;
|
|
|
|
namespace ZB.MOM.WW.MxGateway.Server.Grpc;
|
|
|
|
/// <summary>gRPC service implementation for MXAccess Gateway operations.</summary>
|
|
public sealed class MxAccessGatewayService(
|
|
ISessionManager sessionManager,
|
|
IGatewayRequestIdentityAccessor identityAccessor,
|
|
IConstraintEnforcer constraintEnforcer,
|
|
MxAccessGrpcRequestValidator requestValidator,
|
|
MxAccessGrpcMapper mapper,
|
|
IEventStreamService eventStreamService,
|
|
GatewayMetrics metrics,
|
|
ILogger<MxAccessGatewayService> logger,
|
|
IGatewayAlarmService alarmService) : MxAccessGateway.MxAccessGatewayBase
|
|
{
|
|
/// <inheritdoc />
|
|
public override async Task<OpenSessionReply> OpenSession(
|
|
OpenSessionRequest request,
|
|
ServerCallContext context)
|
|
{
|
|
try
|
|
{
|
|
requestValidator.ValidateOpenSession(request);
|
|
GatewaySession session = await sessionManager
|
|
.OpenSessionAsync(
|
|
SessionOpenRequest.FromContract(request),
|
|
ResolveClientIdentity(),
|
|
identityAccessor.Current?.KeyId,
|
|
context.CancellationToken)
|
|
.ConfigureAwait(false);
|
|
|
|
OpenSessionReply reply = new()
|
|
{
|
|
SessionId = session.SessionId,
|
|
BackendName = session.BackendName,
|
|
WorkerProcessId = session.WorkerProcessId ?? 0,
|
|
WorkerProtocolVersion = GatewayContractInfo.WorkerProtocolVersion,
|
|
GatewayProtocolVersion = GatewayContractInfo.GatewayProtocolVersion,
|
|
DefaultCommandTimeout = Google.Protobuf.WellKnownTypes.Duration.FromTimeSpan(session.CommandTimeout),
|
|
ProtocolStatus = MxAccessGrpcMapper.Ok(),
|
|
};
|
|
reply.Capabilities.Add("unary-open-session");
|
|
reply.Capabilities.Add("unary-close-session");
|
|
reply.Capabilities.Add("unary-invoke");
|
|
reply.Capabilities.Add("server-stream-events");
|
|
reply.Capabilities.Add("bulk-subscribe-commands");
|
|
reply.Capabilities.Add("bulk-read-commands");
|
|
reply.Capabilities.Add("bulk-write-commands");
|
|
reply.Capabilities.Add("unary-acknowledge-alarm");
|
|
reply.Capabilities.Add("server-stream-active-alarms");
|
|
|
|
return reply;
|
|
}
|
|
catch (Exception exception) when (exception is not RpcException)
|
|
{
|
|
throw MapException(exception);
|
|
}
|
|
}
|
|
|
|
/// <inheritdoc />
|
|
public override async Task<CloseSessionReply> CloseSession(
|
|
CloseSessionRequest request,
|
|
ServerCallContext context)
|
|
{
|
|
try
|
|
{
|
|
requestValidator.ValidateCloseSession(request);
|
|
SessionCloseResult result = await sessionManager
|
|
.CloseSessionAsync(request.SessionId, context.CancellationToken)
|
|
.ConfigureAwait(false);
|
|
|
|
return new CloseSessionReply
|
|
{
|
|
SessionId = result.SessionId,
|
|
FinalState = result.FinalState,
|
|
ProtocolStatus = MxAccessGrpcMapper.Ok(result.AlreadyClosed ? "Session was already closed." : "Session closed."),
|
|
};
|
|
}
|
|
catch (Exception exception) when (exception is not RpcException)
|
|
{
|
|
throw MapException(exception);
|
|
}
|
|
}
|
|
|
|
/// <inheritdoc />
|
|
public override async Task<MxCommandReply> Invoke(
|
|
MxCommandRequest request,
|
|
ServerCallContext context)
|
|
{
|
|
try
|
|
{
|
|
requestValidator.ValidateInvoke(request);
|
|
GatewaySession session = ResolveSession(request.SessionId);
|
|
MxCommand command = request.Command;
|
|
BulkConstraintPlan? bulkConstraintPlan = await ApplyConstraintsAsync(
|
|
session,
|
|
command,
|
|
request.ClientCorrelationId,
|
|
context.CancellationToken)
|
|
.ConfigureAwait(false);
|
|
|
|
MxCommand commandToInvoke = bulkConstraintPlan?.Command ?? command;
|
|
if (bulkConstraintPlan is { HasAllowedItems: false })
|
|
{
|
|
return bulkConstraintPlan.CreateDeniedReply(request);
|
|
}
|
|
|
|
MxCommandRequest invokeRequest = request.Clone();
|
|
invokeRequest.Command = commandToInvoke;
|
|
WorkerCommand workerCommand = mapper.MapCommand(invokeRequest);
|
|
WorkerCommandReply workerReply = await sessionManager
|
|
.InvokeAsync(request.SessionId, workerCommand, context.CancellationToken)
|
|
.ConfigureAwait(false);
|
|
|
|
MxCommandReply publicReply = mapper.MapCommandReply(workerReply);
|
|
if (bulkConstraintPlan is not null)
|
|
{
|
|
publicReply = bulkConstraintPlan.MergeDeniedInto(publicReply);
|
|
}
|
|
|
|
session.TrackCommandReply(commandToInvoke, publicReply);
|
|
return publicReply;
|
|
}
|
|
catch (Exception exception) when (exception is not RpcException)
|
|
{
|
|
throw MapException(exception);
|
|
}
|
|
}
|
|
|
|
/// <inheritdoc />
|
|
public override async Task StreamEvents(
|
|
StreamEventsRequest request,
|
|
IServerStreamWriter<MxEvent> responseStream,
|
|
ServerCallContext context)
|
|
{
|
|
try
|
|
{
|
|
requestValidator.ValidateStreamEvents(request);
|
|
await foreach (MxEvent publicEvent in eventStreamService
|
|
.StreamEventsAsync(request, context.CancellationToken)
|
|
.WithCancellation(context.CancellationToken)
|
|
.ConfigureAwait(false))
|
|
{
|
|
Stopwatch stopwatch = Stopwatch.StartNew();
|
|
await responseStream.WriteAsync(publicEvent).ConfigureAwait(false);
|
|
metrics.RecordEventStreamSend(publicEvent.Family.ToString(), stopwatch.Elapsed);
|
|
}
|
|
}
|
|
catch (Exception exception) when (exception is not RpcException)
|
|
{
|
|
throw MapException(exception);
|
|
}
|
|
}
|
|
|
|
/// <inheritdoc />
|
|
/// <remarks>
|
|
/// Surfaces the public AcknowledgeAlarm RPC. Acknowledgement is
|
|
/// session-less: the gateway routes it through the always-on
|
|
/// <see cref="IGatewayAlarmService"/> monitor session. An
|
|
/// <c>alarm_full_reference</c> that parses as a canonical GUID forwards
|
|
/// to <c>AcknowledgeAlarmCommand</c>; a <c>Provider!Group.Tag</c>
|
|
/// reference forwards to <c>AcknowledgeAlarmByNameCommand</c>; anything
|
|
/// else returns an <c>InvalidRequest</c> diagnostic in the reply.
|
|
/// </remarks>
|
|
public override async Task<AcknowledgeAlarmReply> AcknowledgeAlarm(
|
|
AcknowledgeAlarmRequest request,
|
|
ServerCallContext context)
|
|
{
|
|
try
|
|
{
|
|
ArgumentNullException.ThrowIfNull(request);
|
|
if (string.IsNullOrEmpty(request.AlarmFullReference))
|
|
{
|
|
throw new RpcException(new Status(StatusCode.InvalidArgument, "alarm_full_reference is required."));
|
|
}
|
|
|
|
return await alarmService.AcknowledgeAsync(request, context.CancellationToken)
|
|
.ConfigureAwait(false);
|
|
}
|
|
catch (Exception exception) when (exception is not RpcException)
|
|
{
|
|
throw MapException(exception);
|
|
}
|
|
}
|
|
|
|
/// <inheritdoc />
|
|
/// <remarks>
|
|
/// Surfaces the public StreamAlarms RPC — the session-less central
|
|
/// alarm feed. The stream opens with one <c>active_alarm</c> per
|
|
/// currently-active alarm, then a single <c>snapshot_complete</c>, then
|
|
/// a <c>transition</c> for every subsequent change. Served by the
|
|
/// gateway's always-on <see cref="IGatewayAlarmService"/> monitor; any
|
|
/// number of clients fan out from the single monitor.
|
|
/// </remarks>
|
|
public override async Task StreamAlarms(
|
|
StreamAlarmsRequest request,
|
|
IServerStreamWriter<AlarmFeedMessage> responseStream,
|
|
ServerCallContext context)
|
|
{
|
|
try
|
|
{
|
|
ArgumentNullException.ThrowIfNull(request);
|
|
await foreach (AlarmFeedMessage message in alarmService
|
|
.StreamAsync(request.AlarmFilterPrefix, context.CancellationToken)
|
|
.WithCancellation(context.CancellationToken)
|
|
.ConfigureAwait(false))
|
|
{
|
|
await responseStream.WriteAsync(message).ConfigureAwait(false);
|
|
}
|
|
}
|
|
catch (Exception exception) when (exception is not RpcException)
|
|
{
|
|
throw MapException(exception);
|
|
}
|
|
}
|
|
|
|
/// <inheritdoc />
|
|
/// <remarks>
|
|
/// Snapshot of the active-alarm cache maintained by the gateway's
|
|
/// always-on alarm monitor. Streams one <see cref="ActiveAlarmSnapshot"/>
|
|
/// per currently-active alarm and completes — no transitions are
|
|
/// emitted. Use <c>StreamAlarms</c> for a live transition feed.
|
|
/// </remarks>
|
|
public override async Task QueryActiveAlarms(
|
|
QueryActiveAlarmsRequest request,
|
|
IServerStreamWriter<ActiveAlarmSnapshot> responseStream,
|
|
ServerCallContext context)
|
|
{
|
|
try
|
|
{
|
|
ArgumentNullException.ThrowIfNull(request);
|
|
string? filter = string.IsNullOrEmpty(request.AlarmFilterPrefix)
|
|
? null
|
|
: request.AlarmFilterPrefix;
|
|
foreach (ActiveAlarmSnapshot snapshot in alarmService.CurrentAlarms)
|
|
{
|
|
context.CancellationToken.ThrowIfCancellationRequested();
|
|
if (filter is not null
|
|
&& !snapshot.AlarmFullReference.StartsWith(filter, StringComparison.Ordinal))
|
|
{
|
|
continue;
|
|
}
|
|
|
|
await responseStream.WriteAsync(snapshot).ConfigureAwait(false);
|
|
}
|
|
}
|
|
catch (Exception exception) when (exception is not RpcException)
|
|
{
|
|
throw MapException(exception);
|
|
}
|
|
}
|
|
|
|
private string? ResolveClientIdentity()
|
|
{
|
|
return identityAccessor.Current?.DisplayName ?? identityAccessor.Current?.KeyId;
|
|
}
|
|
|
|
private GatewaySession ResolveSession(string sessionId)
|
|
{
|
|
if (!sessionManager.TryGetSession(sessionId, out GatewaySession? session) || session is null)
|
|
{
|
|
throw new SessionManagerException(
|
|
SessionManagerErrorCode.SessionNotFound,
|
|
$"Session {sessionId} was not found.");
|
|
}
|
|
|
|
return session;
|
|
}
|
|
|
|
private async Task<BulkConstraintPlan?> ApplyConstraintsAsync(
|
|
GatewaySession session,
|
|
MxCommand command,
|
|
string? correlationId,
|
|
CancellationToken cancellationToken)
|
|
{
|
|
ApiKeyIdentity? identity = identityAccessor.Current;
|
|
switch (command.Kind)
|
|
{
|
|
case MxCommandKind.AddItem:
|
|
await EnforceReadTagAsync(identity, command.Kind, command.AddItem.ItemDefinition, correlationId, cancellationToken)
|
|
.ConfigureAwait(false);
|
|
return null;
|
|
case MxCommandKind.AddItem2:
|
|
await EnforceReadTagAsync(identity, command.Kind, command.AddItem2.ItemDefinition, correlationId, cancellationToken)
|
|
.ConfigureAwait(false);
|
|
return null;
|
|
case MxCommandKind.AddItemBulk:
|
|
return await FilterTagBulkAsync(
|
|
identity,
|
|
command,
|
|
command.AddItemBulk.ServerHandle,
|
|
command.AddItemBulk.TagAddresses,
|
|
correlationId,
|
|
cancellationToken)
|
|
.ConfigureAwait(false);
|
|
case MxCommandKind.SubscribeBulk:
|
|
return await FilterTagBulkAsync(
|
|
identity,
|
|
command,
|
|
command.SubscribeBulk.ServerHandle,
|
|
command.SubscribeBulk.TagAddresses,
|
|
correlationId,
|
|
cancellationToken)
|
|
.ConfigureAwait(false);
|
|
case MxCommandKind.AdviseItemBulk:
|
|
return await FilterHandleBulkAsync(
|
|
identity,
|
|
session,
|
|
command,
|
|
command.AdviseItemBulk.ServerHandle,
|
|
command.AdviseItemBulk.ItemHandles,
|
|
correlationId,
|
|
cancellationToken)
|
|
.ConfigureAwait(false);
|
|
case MxCommandKind.ReadBulk:
|
|
return await FilterReadBulkAsync(
|
|
identity,
|
|
command,
|
|
command.ReadBulk.ServerHandle,
|
|
command.ReadBulk.TagAddresses,
|
|
correlationId,
|
|
cancellationToken)
|
|
.ConfigureAwait(false);
|
|
case MxCommandKind.WriteBulk:
|
|
return await FilterWriteBulkAsync(
|
|
identity,
|
|
session,
|
|
command,
|
|
command.WriteBulk.ServerHandle,
|
|
command.WriteBulk.Entries,
|
|
entry => entry.ItemHandle,
|
|
correlationId,
|
|
cancellationToken)
|
|
.ConfigureAwait(false);
|
|
case MxCommandKind.Write2Bulk:
|
|
return await FilterWriteBulkAsync(
|
|
identity,
|
|
session,
|
|
command,
|
|
command.Write2Bulk.ServerHandle,
|
|
command.Write2Bulk.Entries,
|
|
entry => entry.ItemHandle,
|
|
correlationId,
|
|
cancellationToken)
|
|
.ConfigureAwait(false);
|
|
case MxCommandKind.WriteSecuredBulk:
|
|
return await FilterWriteBulkAsync(
|
|
identity,
|
|
session,
|
|
command,
|
|
command.WriteSecuredBulk.ServerHandle,
|
|
command.WriteSecuredBulk.Entries,
|
|
entry => entry.ItemHandle,
|
|
correlationId,
|
|
cancellationToken)
|
|
.ConfigureAwait(false);
|
|
case MxCommandKind.WriteSecured2Bulk:
|
|
return await FilterWriteBulkAsync(
|
|
identity,
|
|
session,
|
|
command,
|
|
command.WriteSecured2Bulk.ServerHandle,
|
|
command.WriteSecured2Bulk.Entries,
|
|
entry => entry.ItemHandle,
|
|
correlationId,
|
|
cancellationToken)
|
|
.ConfigureAwait(false);
|
|
case MxCommandKind.Write:
|
|
await EnforceWriteHandleAsync(
|
|
identity,
|
|
session,
|
|
command.Kind,
|
|
command.Write.ServerHandle,
|
|
command.Write.ItemHandle,
|
|
correlationId,
|
|
cancellationToken)
|
|
.ConfigureAwait(false);
|
|
return null;
|
|
case MxCommandKind.Write2:
|
|
await EnforceWriteHandleAsync(
|
|
identity,
|
|
session,
|
|
command.Kind,
|
|
command.Write2.ServerHandle,
|
|
command.Write2.ItemHandle,
|
|
correlationId,
|
|
cancellationToken)
|
|
.ConfigureAwait(false);
|
|
return null;
|
|
case MxCommandKind.WriteSecured:
|
|
await EnforceWriteHandleAsync(
|
|
identity,
|
|
session,
|
|
command.Kind,
|
|
command.WriteSecured.ServerHandle,
|
|
command.WriteSecured.ItemHandle,
|
|
correlationId,
|
|
cancellationToken)
|
|
.ConfigureAwait(false);
|
|
return null;
|
|
case MxCommandKind.WriteSecured2:
|
|
await EnforceWriteHandleAsync(
|
|
identity,
|
|
session,
|
|
command.Kind,
|
|
command.WriteSecured2.ServerHandle,
|
|
command.WriteSecured2.ItemHandle,
|
|
correlationId,
|
|
cancellationToken)
|
|
.ConfigureAwait(false);
|
|
return null;
|
|
default:
|
|
return null;
|
|
}
|
|
}
|
|
|
|
private async Task EnforceReadTagAsync(
|
|
ApiKeyIdentity? identity,
|
|
MxCommandKind commandKind,
|
|
string tagAddress,
|
|
string? correlationId,
|
|
CancellationToken cancellationToken)
|
|
{
|
|
ConstraintFailure? failure = await constraintEnforcer
|
|
.CheckReadTagAsync(identity, tagAddress, cancellationToken)
|
|
.ConfigureAwait(false);
|
|
if (failure is null)
|
|
{
|
|
return;
|
|
}
|
|
|
|
await constraintEnforcer.RecordDenialAsync(identity, commandKind.ToString(), tagAddress, failure, correlationId, cancellationToken)
|
|
.ConfigureAwait(false);
|
|
throw new RpcException(new Status(StatusCode.PermissionDenied, failure.Message));
|
|
}
|
|
|
|
private async Task EnforceWriteHandleAsync(
|
|
ApiKeyIdentity? identity,
|
|
GatewaySession session,
|
|
MxCommandKind commandKind,
|
|
int serverHandle,
|
|
int itemHandle,
|
|
string? correlationId,
|
|
CancellationToken cancellationToken)
|
|
{
|
|
ConstraintFailure? failure = await constraintEnforcer
|
|
.CheckWriteHandleAsync(identity, session, serverHandle, itemHandle, cancellationToken)
|
|
.ConfigureAwait(false);
|
|
if (failure is null)
|
|
{
|
|
return;
|
|
}
|
|
|
|
await constraintEnforcer.RecordDenialAsync(identity, commandKind.ToString(), itemHandle.ToString(System.Globalization.CultureInfo.InvariantCulture), failure, correlationId, cancellationToken)
|
|
.ConfigureAwait(false);
|
|
throw new RpcException(new Status(StatusCode.PermissionDenied, failure.Message));
|
|
}
|
|
|
|
private async Task<BulkConstraintPlan?> FilterTagBulkAsync(
|
|
ApiKeyIdentity? identity,
|
|
MxCommand command,
|
|
int serverHandle,
|
|
IReadOnlyList<string> tagAddresses,
|
|
string? correlationId,
|
|
CancellationToken cancellationToken)
|
|
{
|
|
Dictionary<int, SubscribeResult> denied = [];
|
|
List<string> allowed = [];
|
|
for (int index = 0; index < tagAddresses.Count; index++)
|
|
{
|
|
string tagAddress = tagAddresses[index];
|
|
ConstraintFailure? failure = await constraintEnforcer
|
|
.CheckReadTagAsync(identity, tagAddress, cancellationToken)
|
|
.ConfigureAwait(false);
|
|
if (failure is null)
|
|
{
|
|
allowed.Add(tagAddress);
|
|
continue;
|
|
}
|
|
|
|
await constraintEnforcer.RecordDenialAsync(identity, command.Kind.ToString(), tagAddress, failure, correlationId, cancellationToken)
|
|
.ConfigureAwait(false);
|
|
denied[index] = new SubscribeResult
|
|
{
|
|
ServerHandle = serverHandle,
|
|
TagAddress = tagAddress,
|
|
WasSuccessful = false,
|
|
ErrorMessage = failure.Message,
|
|
};
|
|
}
|
|
|
|
if (denied.Count == 0)
|
|
{
|
|
return null;
|
|
}
|
|
|
|
MxCommand filtered = command.Clone();
|
|
if (filtered.Kind == MxCommandKind.AddItemBulk)
|
|
{
|
|
filtered.AddItemBulk.TagAddresses.Clear();
|
|
filtered.AddItemBulk.TagAddresses.Add(allowed);
|
|
}
|
|
else
|
|
{
|
|
filtered.SubscribeBulk.TagAddresses.Clear();
|
|
filtered.SubscribeBulk.TagAddresses.Add(allowed);
|
|
}
|
|
|
|
return new SubscribeBulkConstraintPlan(filtered, tagAddresses.Count, denied, allowed.Count > 0);
|
|
}
|
|
|
|
private async Task<BulkConstraintPlan?> FilterReadBulkAsync(
|
|
ApiKeyIdentity? identity,
|
|
MxCommand command,
|
|
int serverHandle,
|
|
IReadOnlyList<string> tagAddresses,
|
|
string? correlationId,
|
|
CancellationToken cancellationToken)
|
|
{
|
|
// Mirrors FilterTagBulkAsync but produces BulkReadResult denial entries
|
|
// so the reply payload merges into BulkReadReply.Results, not
|
|
// BulkSubscribeReply.Results.
|
|
Dictionary<int, BulkReadResult> denied = [];
|
|
List<string> allowed = [];
|
|
for (int index = 0; index < tagAddresses.Count; index++)
|
|
{
|
|
string tagAddress = tagAddresses[index];
|
|
ConstraintFailure? failure = await constraintEnforcer
|
|
.CheckReadTagAsync(identity, tagAddress, cancellationToken)
|
|
.ConfigureAwait(false);
|
|
if (failure is null)
|
|
{
|
|
allowed.Add(tagAddress);
|
|
continue;
|
|
}
|
|
|
|
await constraintEnforcer.RecordDenialAsync(identity, command.Kind.ToString(), tagAddress, failure, correlationId, cancellationToken)
|
|
.ConfigureAwait(false);
|
|
denied[index] = new BulkReadResult
|
|
{
|
|
ServerHandle = serverHandle,
|
|
TagAddress = tagAddress,
|
|
WasSuccessful = false,
|
|
WasCached = false,
|
|
ErrorMessage = failure.Message,
|
|
};
|
|
}
|
|
|
|
if (denied.Count == 0)
|
|
{
|
|
return null;
|
|
}
|
|
|
|
MxCommand filtered = command.Clone();
|
|
filtered.ReadBulk.TagAddresses.Clear();
|
|
filtered.ReadBulk.TagAddresses.Add(allowed);
|
|
|
|
return new ReadBulkConstraintPlan(filtered, tagAddresses.Count, denied, allowed.Count > 0);
|
|
}
|
|
|
|
private async Task<BulkConstraintPlan?> FilterWriteBulkAsync<TEntry>(
|
|
ApiKeyIdentity? identity,
|
|
GatewaySession session,
|
|
MxCommand command,
|
|
int serverHandle,
|
|
Google.Protobuf.Collections.RepeatedField<TEntry> entries,
|
|
Func<TEntry, int> getItemHandle,
|
|
string? correlationId,
|
|
CancellationToken cancellationToken) where TEntry : class
|
|
{
|
|
// The four bulk-write families each carry a different per-entry message
|
|
// shape (WriteBulkEntry / Write2BulkEntry / WriteSecuredBulkEntry /
|
|
// WriteSecured2BulkEntry), but the constraint check itself is identical
|
|
// — "is this caller allowed to write to this server+item handle?".
|
|
// Parameterising on TEntry + getItemHandle keeps a single filter
|
|
// routine for all four and avoids duplicating CheckWriteHandleAsync
|
|
// calls.
|
|
Dictionary<int, BulkWriteResult> denied = [];
|
|
List<TEntry> allowed = [];
|
|
for (int index = 0; index < entries.Count; index++)
|
|
{
|
|
TEntry entry = entries[index];
|
|
int itemHandle = getItemHandle(entry);
|
|
ConstraintFailure? failure = await constraintEnforcer
|
|
.CheckWriteHandleAsync(identity, session, serverHandle, itemHandle, cancellationToken)
|
|
.ConfigureAwait(false);
|
|
if (failure is null)
|
|
{
|
|
allowed.Add(entry);
|
|
continue;
|
|
}
|
|
|
|
await constraintEnforcer.RecordDenialAsync(
|
|
identity,
|
|
command.Kind.ToString(),
|
|
itemHandle.ToString(System.Globalization.CultureInfo.InvariantCulture),
|
|
failure,
|
|
correlationId,
|
|
cancellationToken)
|
|
.ConfigureAwait(false);
|
|
denied[index] = new BulkWriteResult
|
|
{
|
|
ServerHandle = serverHandle,
|
|
ItemHandle = itemHandle,
|
|
WasSuccessful = false,
|
|
ErrorMessage = failure.Message,
|
|
};
|
|
}
|
|
|
|
if (denied.Count == 0)
|
|
{
|
|
return null;
|
|
}
|
|
|
|
MxCommand filtered = command.Clone();
|
|
ReplaceWriteBulkEntries(filtered, allowed);
|
|
return new WriteBulkConstraintPlan(filtered, entries.Count, denied, allowed.Count > 0);
|
|
}
|
|
|
|
private static void ReplaceWriteBulkEntries<TEntry>(MxCommand command, IReadOnlyList<TEntry> allowed)
|
|
where TEntry : class
|
|
{
|
|
switch (command.Kind)
|
|
{
|
|
case MxCommandKind.WriteBulk:
|
|
command.WriteBulk.Entries.Clear();
|
|
command.WriteBulk.Entries.Add((IEnumerable<WriteBulkEntry>)allowed);
|
|
break;
|
|
case MxCommandKind.Write2Bulk:
|
|
command.Write2Bulk.Entries.Clear();
|
|
command.Write2Bulk.Entries.Add((IEnumerable<Write2BulkEntry>)allowed);
|
|
break;
|
|
case MxCommandKind.WriteSecuredBulk:
|
|
command.WriteSecuredBulk.Entries.Clear();
|
|
command.WriteSecuredBulk.Entries.Add((IEnumerable<WriteSecuredBulkEntry>)allowed);
|
|
break;
|
|
case MxCommandKind.WriteSecured2Bulk:
|
|
command.WriteSecured2Bulk.Entries.Clear();
|
|
command.WriteSecured2Bulk.Entries.Add((IEnumerable<WriteSecured2BulkEntry>)allowed);
|
|
break;
|
|
}
|
|
}
|
|
|
|
private async Task<BulkConstraintPlan?> FilterHandleBulkAsync(
|
|
ApiKeyIdentity? identity,
|
|
GatewaySession session,
|
|
MxCommand command,
|
|
int serverHandle,
|
|
IReadOnlyList<int> itemHandles,
|
|
string? correlationId,
|
|
CancellationToken cancellationToken)
|
|
{
|
|
Dictionary<int, SubscribeResult> denied = [];
|
|
List<int> allowed = [];
|
|
for (int index = 0; index < itemHandles.Count; index++)
|
|
{
|
|
int itemHandle = itemHandles[index];
|
|
ConstraintFailure? failure = await constraintEnforcer
|
|
.CheckReadHandleAsync(identity, session, serverHandle, itemHandle, cancellationToken)
|
|
.ConfigureAwait(false);
|
|
if (failure is null)
|
|
{
|
|
allowed.Add(itemHandle);
|
|
continue;
|
|
}
|
|
|
|
await constraintEnforcer.RecordDenialAsync(identity, command.Kind.ToString(), itemHandle.ToString(System.Globalization.CultureInfo.InvariantCulture), failure, correlationId, cancellationToken)
|
|
.ConfigureAwait(false);
|
|
denied[index] = new SubscribeResult
|
|
{
|
|
ServerHandle = serverHandle,
|
|
ItemHandle = itemHandle,
|
|
WasSuccessful = false,
|
|
ErrorMessage = failure.Message,
|
|
};
|
|
}
|
|
|
|
if (denied.Count == 0)
|
|
{
|
|
return null;
|
|
}
|
|
|
|
MxCommand filtered = command.Clone();
|
|
filtered.AdviseItemBulk.ItemHandles.Clear();
|
|
filtered.AdviseItemBulk.ItemHandles.Add(allowed);
|
|
|
|
return new SubscribeBulkConstraintPlan(filtered, itemHandles.Count, denied, allowed.Count > 0);
|
|
}
|
|
|
|
/// <summary>
|
|
/// Polymorphic constraint plan returned from <see cref="ApplyConstraintsAsync"/>.
|
|
/// Each concrete subtype is keyed to a specific bulk-reply shape — the
|
|
/// SubscribeResult-based AddItem/Advise/Subscribe family, the
|
|
/// BulkWriteResult-based Write* bulk family, and the BulkReadResult-based
|
|
/// ReadBulk command. Subtypes own their own merge / denied-reply build
|
|
/// logic so the Invoke dispatch site never branches on reply shape.
|
|
/// </summary>
|
|
private abstract record BulkConstraintPlan(
|
|
MxCommand Command,
|
|
int OriginalCount,
|
|
bool HasAllowedItems)
|
|
{
|
|
/// <summary>Builds a reply containing only the denied entries (used when no items survived filtering).</summary>
|
|
/// <param name="request">The original command request.</param>
|
|
/// <returns>A reply with denied results only.</returns>
|
|
public abstract MxCommandReply CreateDeniedReply(MxCommandRequest request);
|
|
|
|
/// <summary>Splices denied entries back into the worker's allowed-only reply in original-index order.</summary>
|
|
/// <param name="reply">The worker's reply containing only allowed results.</param>
|
|
/// <returns>A merged reply in the original order.</returns>
|
|
public abstract MxCommandReply MergeDeniedInto(MxCommandReply reply);
|
|
}
|
|
|
|
private sealed record SubscribeBulkConstraintPlan(
|
|
MxCommand Command,
|
|
int OriginalCount,
|
|
IReadOnlyDictionary<int, SubscribeResult> DeniedResults,
|
|
bool HasAllowedItems)
|
|
: BulkConstraintPlan(Command, OriginalCount, HasAllowedItems)
|
|
{
|
|
/// <inheritdoc />
|
|
public override MxCommandReply CreateDeniedReply(MxCommandRequest request)
|
|
{
|
|
MxCommandReply reply = new()
|
|
{
|
|
SessionId = request.SessionId,
|
|
CorrelationId = request.ClientCorrelationId,
|
|
Kind = request.Command.Kind,
|
|
ProtocolStatus = MxAccessGrpcMapper.Ok(),
|
|
};
|
|
SetPayload(reply, BuildMerged(new BulkSubscribeReply()));
|
|
return reply;
|
|
}
|
|
|
|
/// <inheritdoc />
|
|
public override MxCommandReply MergeDeniedInto(MxCommandReply reply)
|
|
{
|
|
BulkSubscribeReply allowed = GetPayload(reply) ?? new BulkSubscribeReply();
|
|
SetPayload(reply, BuildMerged(allowed));
|
|
return reply;
|
|
}
|
|
|
|
private BulkSubscribeReply BuildMerged(BulkSubscribeReply allowed)
|
|
{
|
|
Queue<SubscribeResult> allowedResults = new(allowed.Results);
|
|
BulkSubscribeReply merged = new();
|
|
for (int index = 0; index < OriginalCount; index++)
|
|
{
|
|
if (DeniedResults.TryGetValue(index, out SubscribeResult? denied))
|
|
{
|
|
merged.Results.Add(denied);
|
|
}
|
|
else if (allowedResults.TryDequeue(out SubscribeResult? allowedResult))
|
|
{
|
|
merged.Results.Add(allowedResult);
|
|
}
|
|
}
|
|
|
|
return merged;
|
|
}
|
|
|
|
private BulkSubscribeReply? GetPayload(MxCommandReply reply) => Command.Kind switch
|
|
{
|
|
MxCommandKind.AddItemBulk => reply.AddItemBulk,
|
|
MxCommandKind.AdviseItemBulk => reply.AdviseItemBulk,
|
|
MxCommandKind.SubscribeBulk => reply.SubscribeBulk,
|
|
_ => null,
|
|
};
|
|
|
|
private void SetPayload(MxCommandReply reply, BulkSubscribeReply payload)
|
|
{
|
|
switch (Command.Kind)
|
|
{
|
|
case MxCommandKind.AddItemBulk:
|
|
reply.AddItemBulk = payload;
|
|
break;
|
|
case MxCommandKind.AdviseItemBulk:
|
|
reply.AdviseItemBulk = payload;
|
|
break;
|
|
case MxCommandKind.SubscribeBulk:
|
|
reply.SubscribeBulk = payload;
|
|
break;
|
|
}
|
|
}
|
|
}
|
|
|
|
private sealed record WriteBulkConstraintPlan(
|
|
MxCommand Command,
|
|
int OriginalCount,
|
|
IReadOnlyDictionary<int, BulkWriteResult> DeniedResults,
|
|
bool HasAllowedItems)
|
|
: BulkConstraintPlan(Command, OriginalCount, HasAllowedItems)
|
|
{
|
|
/// <inheritdoc />
|
|
public override MxCommandReply CreateDeniedReply(MxCommandRequest request)
|
|
{
|
|
MxCommandReply reply = new()
|
|
{
|
|
SessionId = request.SessionId,
|
|
CorrelationId = request.ClientCorrelationId,
|
|
Kind = request.Command.Kind,
|
|
ProtocolStatus = MxAccessGrpcMapper.Ok(),
|
|
};
|
|
SetPayload(reply, BuildMerged(new BulkWriteReply()));
|
|
return reply;
|
|
}
|
|
|
|
/// <inheritdoc />
|
|
public override MxCommandReply MergeDeniedInto(MxCommandReply reply)
|
|
{
|
|
BulkWriteReply allowed = GetPayload(reply) ?? new BulkWriteReply();
|
|
SetPayload(reply, BuildMerged(allowed));
|
|
return reply;
|
|
}
|
|
|
|
private BulkWriteReply BuildMerged(BulkWriteReply allowed)
|
|
{
|
|
Queue<BulkWriteResult> allowedResults = new(allowed.Results);
|
|
BulkWriteReply merged = new();
|
|
for (int index = 0; index < OriginalCount; index++)
|
|
{
|
|
if (DeniedResults.TryGetValue(index, out BulkWriteResult? denied))
|
|
{
|
|
merged.Results.Add(denied);
|
|
}
|
|
else if (allowedResults.TryDequeue(out BulkWriteResult? allowedResult))
|
|
{
|
|
merged.Results.Add(allowedResult);
|
|
}
|
|
}
|
|
|
|
return merged;
|
|
}
|
|
|
|
private BulkWriteReply? GetPayload(MxCommandReply reply) => Command.Kind switch
|
|
{
|
|
MxCommandKind.WriteBulk => reply.WriteBulk,
|
|
MxCommandKind.Write2Bulk => reply.Write2Bulk,
|
|
MxCommandKind.WriteSecuredBulk => reply.WriteSecuredBulk,
|
|
MxCommandKind.WriteSecured2Bulk => reply.WriteSecured2Bulk,
|
|
_ => null,
|
|
};
|
|
|
|
private void SetPayload(MxCommandReply reply, BulkWriteReply payload)
|
|
{
|
|
switch (Command.Kind)
|
|
{
|
|
case MxCommandKind.WriteBulk:
|
|
reply.WriteBulk = payload;
|
|
break;
|
|
case MxCommandKind.Write2Bulk:
|
|
reply.Write2Bulk = payload;
|
|
break;
|
|
case MxCommandKind.WriteSecuredBulk:
|
|
reply.WriteSecuredBulk = payload;
|
|
break;
|
|
case MxCommandKind.WriteSecured2Bulk:
|
|
reply.WriteSecured2Bulk = payload;
|
|
break;
|
|
}
|
|
}
|
|
}
|
|
|
|
private sealed record ReadBulkConstraintPlan(
|
|
MxCommand Command,
|
|
int OriginalCount,
|
|
IReadOnlyDictionary<int, BulkReadResult> DeniedResults,
|
|
bool HasAllowedItems)
|
|
: BulkConstraintPlan(Command, OriginalCount, HasAllowedItems)
|
|
{
|
|
/// <inheritdoc />
|
|
public override MxCommandReply CreateDeniedReply(MxCommandRequest request)
|
|
{
|
|
MxCommandReply reply = new()
|
|
{
|
|
SessionId = request.SessionId,
|
|
CorrelationId = request.ClientCorrelationId,
|
|
Kind = request.Command.Kind,
|
|
ProtocolStatus = MxAccessGrpcMapper.Ok(),
|
|
};
|
|
reply.ReadBulk = BuildMerged(new BulkReadReply());
|
|
return reply;
|
|
}
|
|
|
|
/// <inheritdoc />
|
|
public override MxCommandReply MergeDeniedInto(MxCommandReply reply)
|
|
{
|
|
BulkReadReply allowed = reply.ReadBulk ?? new BulkReadReply();
|
|
reply.ReadBulk = BuildMerged(allowed);
|
|
return reply;
|
|
}
|
|
|
|
private BulkReadReply BuildMerged(BulkReadReply allowed)
|
|
{
|
|
Queue<BulkReadResult> allowedResults = new(allowed.Results);
|
|
BulkReadReply merged = new();
|
|
for (int index = 0; index < OriginalCount; index++)
|
|
{
|
|
if (DeniedResults.TryGetValue(index, out BulkReadResult? denied))
|
|
{
|
|
merged.Results.Add(denied);
|
|
}
|
|
else if (allowedResults.TryDequeue(out BulkReadResult? allowedResult))
|
|
{
|
|
merged.Results.Add(allowedResult);
|
|
}
|
|
}
|
|
|
|
return merged;
|
|
}
|
|
}
|
|
|
|
private RpcException MapException(Exception exception)
|
|
{
|
|
if (exception is OperationCanceledException)
|
|
{
|
|
return new RpcException(new Status(StatusCode.Cancelled, "gRPC request was canceled."));
|
|
}
|
|
|
|
if (exception is SessionManagerException sessionException)
|
|
{
|
|
return MapSessionException(sessionException);
|
|
}
|
|
|
|
if (exception is WorkerClientException workerClientException)
|
|
{
|
|
return MapWorkerClientException(workerClientException);
|
|
}
|
|
|
|
logger.LogWarning(exception, "Public gRPC request failed.");
|
|
return new RpcException(new Status(StatusCode.Unavailable, "Gateway request failed before an MXAccess reply was available."));
|
|
}
|
|
|
|
private static RpcException MapSessionException(SessionManagerException exception)
|
|
{
|
|
StatusCode statusCode = exception.ErrorCode switch
|
|
{
|
|
SessionManagerErrorCode.SessionNotFound => StatusCode.NotFound,
|
|
SessionManagerErrorCode.SessionNotReady => StatusCode.FailedPrecondition,
|
|
SessionManagerErrorCode.EventSubscriberAlreadyActive => StatusCode.ResourceExhausted,
|
|
SessionManagerErrorCode.EventQueueOverflow => StatusCode.ResourceExhausted,
|
|
SessionManagerErrorCode.SessionLimitExceeded => StatusCode.ResourceExhausted,
|
|
SessionManagerErrorCode.OpenFailed => StatusCode.Unavailable,
|
|
SessionManagerErrorCode.CloseFailed => StatusCode.Unavailable,
|
|
_ => StatusCode.Unavailable,
|
|
};
|
|
|
|
return new RpcException(new Status(statusCode, exception.Message));
|
|
}
|
|
|
|
private static RpcException MapWorkerClientException(WorkerClientException exception)
|
|
{
|
|
StatusCode statusCode = exception.ErrorCode switch
|
|
{
|
|
WorkerClientErrorCode.CommandTimeout => StatusCode.DeadlineExceeded,
|
|
WorkerClientErrorCode.GatewayShutdown => StatusCode.Cancelled,
|
|
WorkerClientErrorCode.InvalidState => StatusCode.FailedPrecondition,
|
|
WorkerClientErrorCode.ProtocolViolation => StatusCode.Internal,
|
|
_ => StatusCode.Unavailable,
|
|
};
|
|
|
|
return new RpcException(new Status(statusCode, exception.Message));
|
|
}
|
|
}
|