From c0ccfa27fb8d0dd31fb9cbda15c17e1c2ade6466 Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Thu, 6 Aug 2026 02:53:50 -0400 Subject: [PATCH 01/12] feat(mesh): mesh_transport.v1 proto contract --- .../Protos/MeshTransportProtoContract.cs | 33 +++++ .../Protos/mesh_transport.proto | 118 ++++++++++++++++++ .../ZB.MOM.WW.OtOpcUa.Commons.csproj | 1 + .../MeshTransportProtoContractTests.cs | 49 ++++++++ 4 files changed, 201 insertions(+) create mode 100644 src/Core/ZB.MOM.WW.OtOpcUa.Commons/Protos/MeshTransportProtoContract.cs create mode 100644 src/Core/ZB.MOM.WW.OtOpcUa.Commons/Protos/mesh_transport.proto create mode 100644 tests/Core/ZB.MOM.WW.OtOpcUa.Commons.Tests/MeshTransportProtoContractTests.cs diff --git a/src/Core/ZB.MOM.WW.OtOpcUa.Commons/Protos/MeshTransportProtoContract.cs b/src/Core/ZB.MOM.WW.OtOpcUa.Commons/Protos/MeshTransportProtoContract.cs new file mode 100644 index 00000000..d49ca315 --- /dev/null +++ b/src/Core/ZB.MOM.WW.OtOpcUa.Commons/Protos/MeshTransportProtoContract.cs @@ -0,0 +1,33 @@ +using ZB.MOM.WW.OtOpcUa.Commons.Protos.MeshTransport.V1; + +namespace ZB.MOM.WW.OtOpcUa.Commons.Protos; + +/// +/// The single source of truth for which variants +/// the cross-mesh command transport actually handles. Both the contract-lock test (which proves +/// this list stays in lock-step with the generated oneof) and the oneof↔domain mapper reference +/// this array, so adding a fifth command to mesh_transport.proto forces a matching entry +/// here or the build/test goes red. +/// +/// +/// Mirrors , and for a sharper reason: a telemetry variant +/// nobody handles loses an observation, whereas a command variant nobody handles loses a +/// deployment, a driver restart or an alarm acknowledgement — silently, since the sender's ack +/// only ever meant "received". +/// +public static class MeshTransportProtoContract +{ + /// + /// Exactly the four cross-mesh command oneof cases mirrored from the domain records: + /// DispatchDeployment / RestartDriver / ReconnectDriver / AlarmCommand. The fifth message on + /// this contract, ApplyAckMessage, travels the opposite direction on its own RPC and is + /// not a oneof member, so it is deliberately absent from this list. + /// + public static readonly CommandEnvelope.CommandOneofCase[] HandledCases = + [ + CommandEnvelope.CommandOneofCase.DispatchDeployment, + CommandEnvelope.CommandOneofCase.RestartDriver, + CommandEnvelope.CommandOneofCase.ReconnectDriver, + CommandEnvelope.CommandOneofCase.AlarmCommand, + ]; +} diff --git a/src/Core/ZB.MOM.WW.OtOpcUa.Commons/Protos/mesh_transport.proto b/src/Core/ZB.MOM.WW.OtOpcUa.Commons/Protos/mesh_transport.proto new file mode 100644 index 00000000..d91c52f8 --- /dev/null +++ b/src/Core/ZB.MOM.WW.OtOpcUa.Commons/Protos/mesh_transport.proto @@ -0,0 +1,118 @@ +syntax = "proto3"; + +package mesh_transport.v1; + +option csharp_namespace = "ZB.MOM.WW.OtOpcUa.Commons.Protos.MeshTransport.V1"; + +import "google/protobuf/timestamp.proto"; + +// ClusterClient→gRPC migration Phase 0: the cross-mesh command plane. The five messages that cross +// an application-Cluster boundary today ride an Akka ClusterClient receptionist; this contract is +// the replacement carrier. Four are central→node commands, one is the node→central deployment ack. +// +// The C# namespace deliberately says "MeshTransport", never "MeshCommand" — the latter is already +// the CLR envelope record ZB.MOM.WW.OtOpcUa.Commons.Messages.Mesh.MeshCommand, and a generated type +// of the same name would collide on every file that touches both sides of the mapper. +// +// Additive-only field evolution: never renumber or reuse a tag; a pre-field peer must read a new +// field's proto3 default correctly (which is why nullable domain members are modelled with `optional` +// for explicit presence, and why every enum reserves 0 as UNSPECIFIED). The oneof is locked by +// MeshTransportProtoContract + its test in the Commons test project. + +// Central → node. Delivered to the node's NodeCommunicationActor, which re-emits the decoded message +// on the node's EventStream. ONE-WAY semantics, preserved verbatim from the ClusterClient transport: +// the ack means "received and re-emitted", nothing more — not applied, not acknowledged by any +// subscriber. No retry is layered on a failure; a failed Execute is logged per node and dropped, +// which is an honest upgrade over the receptionist's silent drop rather than a semantic change. +service NodeCommandService { + rpc Execute (CommandEnvelope) returns (ExecuteAck); +} + +// Node → central. Delivered to central's CentralCommunicationActor, which forwards to the +// ConfigPublishCoordinator singleton proxy — so an ack landing on either node of the central pair +// routes identically, exactly as the receptionist made arrival-node-agnostic. ONE-WAY semantics: the +// ack means "handed to the coordinator", not "the deployment is complete". +service CentralCommandService { + rpc ReportApply (ApplyAckMessage) returns (ReportAck); +} + +message CommandEnvelope { + // Transport-level correlation, carried for every command including AlarmCommandMessage, which is + // the one CLR record with no correlation member of its own. The other three ALSO carry their own + // domain correlation field-for-field below; the duplication is deliberate, so each command stays + // independently round-trippable through the mapper and the envelope stays loggable on its own. + string correlation_id = 1; + oneof command { + DispatchDeploymentCommand dispatch_deployment = 2; // <- Messages.Deploy.DispatchDeployment + RestartDriverCommand restart_driver = 3; // <- Messages.Admin.RestartDriver + ReconnectDriverCommand reconnect_driver = 4; // <- Messages.Admin.ReconnectDriver + AlarmCommandMessage alarm_command = 5; // <- OpcUa.AlarmCommand + } +} + +// Deliberately empty. A non-OK gRPC status is the ONLY failure signal on this call — putting a +// success flag in the body would invite a caller to read OK+false as a distinct outcome the +// fire-and-forget contract does not have. Extensible later without a version bump. +message ExecuteAck { +} + +// Deliberately empty, for the same reason as ExecuteAck. +message ReportAck { +} + +// Mirrors ZB.MOM.WW.OtOpcUa.Commons.Messages.Deploy.DispatchDeployment. +message DispatchDeploymentCommand { + string deployment_id = 1; // DeploymentId (Guid) in its "N" hex form + string revision_hash = 2; // RevisionHash — lowercase 64-char SHA-256 hex + string correlation_id = 3; // CorrelationId (Guid) in its "N" hex form +} + +// Mirrors ZB.MOM.WW.OtOpcUa.Commons.Messages.Admin.RestartDriver. +message RestartDriverCommand { + string cluster_id = 1; + string driver_instance_id = 2; + string actor_by_user_name = 3; + string correlation_id = 4; // Guid in the record; "N" hex form +} + +// Mirrors ZB.MOM.WW.OtOpcUa.Commons.Messages.Admin.ReconnectDriver. +message ReconnectDriverCommand { + string cluster_id = 1; + string driver_instance_id = 2; + string actor_by_user_name = 3; + string correlation_id = 4; // Guid in the record; "N" hex form +} + +// Mirrors ZB.MOM.WW.OtOpcUa.Commons.OpcUa.AlarmCommand. +message AlarmCommandMessage { + string alarm_id = 1; + // The Part 9 operation (Acknowledge / Confirm / OneShotShelve / TimedShelve / Unshelve / Enable / + // Disable / AddComment). A `string` in the CLR record, and kept as one here — the same call the + // telemetry contract makes for transition_kind / state / level. Promoting it to a proto enum would + // put a lossy translation in the mapper for a set the CLR side does not constrain: an operation the + // enum did not know would decode to UNSPECIFIED and the command would be dropped, which is the + // exact silent-drop class this migration exists to remove. + string operation = 2; + string user = 3; + optional string comment = 4; // nullable in the record — presence distinguishes null from "" + google.protobuf.Timestamp unshelve_at_utc = 5; // DateTime? — absent Timestamp encodes null +} + +// Mirrors ZB.MOM.WW.OtOpcUa.Commons.Messages.Deploy.ApplyAck. +message ApplyAckMessage { + string deployment_id = 1; // DeploymentId (Guid) in its "N" hex form + string node_id = 2; // NodeId — a logical host name, already a string + ApplyOutcome outcome = 3; + optional string failure_reason = 4; // nullable in the record — presence distinguishes null from "" + string correlation_id = 5; // CorrelationId (Guid) in its "N" hex form +} + +// Mirrors ZB.MOM.WW.OtOpcUa.Commons.Messages.Deploy.ApplyAckOutcome, shifted by one so 0 stays +// meaningless. proto3 decodes an absent enum field as 0; if 0 were Applied, an ack that lost the +// field would report a deployment applied when the node said nothing at all — and the deploy path +// seals on exactly that signal. +enum ApplyOutcome { + APPLY_OUTCOME_UNSPECIFIED = 0; + APPLY_OUTCOME_APPLIED = 1; + APPLY_OUTCOME_FAILED = 2; +} diff --git a/src/Core/ZB.MOM.WW.OtOpcUa.Commons/ZB.MOM.WW.OtOpcUa.Commons.csproj b/src/Core/ZB.MOM.WW.OtOpcUa.Commons/ZB.MOM.WW.OtOpcUa.Commons.csproj index 5e55544f..bd560b43 100644 --- a/src/Core/ZB.MOM.WW.OtOpcUa.Commons/ZB.MOM.WW.OtOpcUa.Commons.csproj +++ b/src/Core/ZB.MOM.WW.OtOpcUa.Commons/ZB.MOM.WW.OtOpcUa.Commons.csproj @@ -28,6 +28,7 @@ + diff --git a/tests/Core/ZB.MOM.WW.OtOpcUa.Commons.Tests/MeshTransportProtoContractTests.cs b/tests/Core/ZB.MOM.WW.OtOpcUa.Commons.Tests/MeshTransportProtoContractTests.cs new file mode 100644 index 00000000..a7c33c08 --- /dev/null +++ b/tests/Core/ZB.MOM.WW.OtOpcUa.Commons.Tests/MeshTransportProtoContractTests.cs @@ -0,0 +1,49 @@ +using Shouldly; +using Xunit; +using ZB.MOM.WW.OtOpcUa.Commons.Protos; +using ZB.MOM.WW.OtOpcUa.Commons.Protos.MeshTransport.V1; + +namespace ZB.MOM.WW.OtOpcUa.Commons.Tests; + +/// +/// Contract-lock for the ClusterClient→gRPC command envelope's oneof. If a fifth command is added +/// to mesh_transport.proto without adding a matching entry to +/// , this test goes red — the single guard +/// that keeps the wire contract and the handled-case set from silently drifting apart. A drift +/// here is exactly the failure the whole transport exists to avoid: a command that crosses the +/// mesh boundary, decodes to a case nobody handles, and is dropped without a trace. +/// +public class MeshTransportProtoContractTests +{ + [Fact] + public void EveryOneofVariant_IsAccountedFor() + { + var variants = System.Enum.GetValues() + .Where(c => c != CommandEnvelope.CommandOneofCase.None).ToArray(); + + variants.ShouldBe(MeshTransportProtoContract.HandledCases, ignoreOrder: true); + } + + [Fact] + public void The_service_base_and_client_types_generate() + { + // Referencing these types is the whole assertion — GrpcServices="Both" must emit both halves + // of both services: central hosts CentralCommandService and dials NodeCommandService, and a + // driver node does the mirror image, so neither service can ship client-only or server-only. + typeof(NodeCommandService.NodeCommandServiceBase).ShouldNotBeNull(); + typeof(NodeCommandService.NodeCommandServiceClient).ShouldNotBeNull(); + typeof(CentralCommandService.CentralCommandServiceBase).ShouldNotBeNull(); + typeof(CentralCommandService.CentralCommandServiceClient).ShouldNotBeNull(); + } + + [Fact] + public void The_apply_outcome_enum_reserves_zero_as_unspecified() + { + // proto3 decodes an absent enum field as 0. If 0 meant "Applied", a message that lost the + // field in transit — or was written by a peer that predates it — would report a deployment + // as successfully applied when nothing was said at all. 0 must stay meaningless. + ((int)ApplyOutcome.Unspecified).ShouldBe(0); + ((int)ApplyOutcome.Applied).ShouldNotBe(0); + ((int)ApplyOutcome.Failed).ShouldNotBe(0); + } +} From aa21f61fa051b9d19cc892a116065d7f7792af40 Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Thu, 6 Aug 2026 03:00:59 -0400 Subject: [PATCH 02/12] feat(mesh): Grpc mode value + validation --- .../MeshTransportOptions.cs | 117 +++++++- .../MeshTransportOptionsValidator.cs | 96 ++++++- .../SplitTopologyTransportValidator.cs | 12 +- .../AdminOperations/AdminOperationsActor.cs | 8 +- .../Coordinators/ConfigPublishCoordinator.cs | 7 +- .../MeshTransportOptionsValidatorTests.cs | 253 +++++++++++++++++- .../SplitTopologyTransportValidatorTests.cs | 49 ++++ 7 files changed, 521 insertions(+), 21 deletions(-) diff --git a/src/Core/ZB.MOM.WW.OtOpcUa.Cluster/MeshTransportOptions.cs b/src/Core/ZB.MOM.WW.OtOpcUa.Cluster/MeshTransportOptions.cs index d3043e0e..d3e41863 100644 --- a/src/Core/ZB.MOM.WW.OtOpcUa.Cluster/MeshTransportOptions.cs +++ b/src/Core/ZB.MOM.WW.OtOpcUa.Cluster/MeshTransportOptions.cs @@ -6,11 +6,18 @@ namespace ZB.MOM.WW.OtOpcUa.Cluster; /// /// /// -/// is a deliberate dark switch, not a permanent knob: both -/// transports live in the tree for one phase so the docker-dev rig can run the same -/// deployment over each and compare, and so a bad Phase 2 rolls back with a config change -/// rather than a revert. Phase 6 deletes the DistributedPubSub branch along with the single -/// mesh that made it viable. +/// is a deliberate dark switch: the transports live in the tree +/// side by side so the docker-dev rig can run the same deployment over each and compare, and +/// so a bad rollout rolls back with a config change rather than a revert. +/// +/// +/// The DistributedPubSub branch is retained, not scheduled for deletion. An earlier +/// comment here promised Phase 6 would delete it along with the single mesh; that was +/// deliberately reversed — see docs/Configuration.md § "The compiled transport-mode +/// defaults were deliberately NOT flipped". DPS remains the correct and default transport for +/// a single-mesh topology, which every integration test and every profile that leaves this +/// key unset still runs. What is being replaced is , by +/// . /// /// /// Discovery is deliberately asymmetric, matching the sister project. Central @@ -32,9 +39,19 @@ public sealed class MeshTransportOptions public const string ModeClusterClient = "ClusterClient"; /// - /// (default) or . Any other value fails - /// the host at startup rather than silently falling back — a transport that is neither of - /// these delivers nothing, and the symptom would be commands that vanish. + /// Commands cross the mesh boundary as unary gRPC calls on the mesh_transport.v1 + /// contract — central dials each node's NodeCommandService, a node posts its + /// deployment ack to central's CentralCommandService. The replacement for + /// ; unlike it, the command plane is authenticated + /// (). + /// + public const string ModeGrpc = "Grpc"; + + /// + /// (default), or + /// . Any other value fails the host at startup rather than silently + /// falling back — a transport that is none of these delivers nothing, and the symptom would + /// be commands that vanish. /// public string Mode { get; set; } = ModeDps; @@ -58,4 +75,88 @@ public sealed class MeshTransportOptions /// process — migration, direct SQL, a second admin node — is picked up without a restart. /// public int ContactRefreshSeconds { get; set; } = 60; + + /// + /// MeshTransport:CentralGrpcEndpoints — node-side only, and only under + /// : the gRPC base addresses of the central pair's + /// CentralCommandService, e.g. http://central-1:4055. Full URLs + /// (scheme://host:port), not bare host:port — http:// is h2c. + /// + /// + /// Tried in order, stopping at the first success, which is what preserves the + /// receptionist's one-ack-per-node-per-dispatch semantic: either central node routes the ack + /// to the same ConfigPublishCoordinator singleton proxy, so which one answers does not + /// matter, but two answering would double-count. The gRPC counterpart of + /// , and it replaces it — a node running + /// never reads the Akka contact points. + /// + public string[] CentralGrpcEndpoints { get; set; } = []; + + /// + /// MeshTransport:GrpcPsk — this node's pre-shared key for the command plane, one per + /// application Cluster. Under it is both the bearer this node + /// presents when posting an ack to central and the bearer it requires on inbound + /// NodeCommandService calls. + /// + /// + /// + /// Production form is a secret reference — ${secret:OT-MESH-PSK-<clusterId>}, + /// e.g. ${secret:OT-MESH-PSK-SITE-A} — resolved pre-host by the existing + /// SecretReferenceExpander, so it works on a driver-only node with no ConfigDb. + /// Supply a literal only on a rig, via the environment + /// (MeshTransport__GrpcPsk); never commit one. + /// + /// + /// Per-Cluster, not fleet-wide, on purpose. The three existing shared-bearer gRPC + /// channels are read/telemetry planes; this one carries RestartDriver, + /// AlarmCommand and deployment dispatch, so a single fleet key would let one + /// compromised site node command every other site and forge acks fleet-wide. + /// + /// + /// Nullable because unset must be distinguishable from blank at the boundary that + /// fails closed: the serve-side interceptor rejects every call when no key is + /// configured rather than defaulting to an empty comparison that any empty bearer would + /// satisfy. + /// + /// + public string? GrpcPsk { get; set; } + + /// + /// MeshTransport:DispatchTimeoutSeconds — central-side deadline, in seconds, for one + /// NodeCommandService.Execute call to one node. Must be greater than 0 under + /// . + /// + /// + /// Bounded per node because a fan-out dispatch dials every enabled node in the cluster: an + /// unbounded call to one unreachable node would stall the commands queued behind it. Expiry + /// is logged and dropped — the transport is fire-and-forget and layers no retry, so the + /// deploy path's 2-minute apply deadline stays the only recovery mechanism. + /// + public int DispatchTimeoutSeconds { get; set; } = 10; + + /// + /// MeshTransport:AckTimeoutSeconds — node-side deadline, in seconds, for one + /// CentralCommandService.ReportApply call to one central endpoint. Must be greater + /// than 0 under . + /// + /// + /// Applied per endpoint, not per ack: the node walks in + /// order, so a dead first endpoint costs this deadline before the second is tried. + /// + public int AckTimeoutSeconds { get; set; } = 10; + + /// + /// MeshTransport:ClusterPsks:{clusterId} — central-side override map of per-cluster + /// PSKs, consulted before the Secrets store when central builds a channel to (or + /// verifies an ack from) a given Cluster. Keyed case-insensitively. + /// + /// + /// Production leaves this empty and the runtime provider resolves + /// OT-MESH-PSK-{clusterId} from the Secrets store instead; the map exists so the + /// docker-dev rig and integration tests can supply literals through the environment + /// (MeshTransport__ClusterPsks__SITE-A) without standing up a KEK. Central needs a map + /// rather than a single key because its cluster set is DB-dynamic — one key per site pair it + /// talks to — whereas a node holds exactly its own (). + /// + public Dictionary ClusterPsks { get; set; } = new(StringComparer.OrdinalIgnoreCase); } diff --git a/src/Core/ZB.MOM.WW.OtOpcUa.Cluster/MeshTransportOptionsValidator.cs b/src/Core/ZB.MOM.WW.OtOpcUa.Cluster/MeshTransportOptionsValidator.cs index afcd8779..ad47fc88 100644 --- a/src/Core/ZB.MOM.WW.OtOpcUa.Cluster/MeshTransportOptionsValidator.cs +++ b/src/Core/ZB.MOM.WW.OtOpcUa.Cluster/MeshTransportOptionsValidator.cs @@ -1,3 +1,4 @@ +using Microsoft.Extensions.Configuration; using Microsoft.Extensions.Options; namespace ZB.MOM.WW.OtOpcUa.Cluster; @@ -15,6 +16,22 @@ namespace ZB.MOM.WW.OtOpcUa.Cluster; /// public sealed class MeshTransportOptionsValidator : IValidateOptions { + private readonly IConfiguration _configuration; + + /// + /// DI-constructed by AddValidatedOptions (a plain AddSingleton), so a + /// constructor dependency is safe here. — not + /// IClusterRoleInfo — is the source of this node's roles: IClusterRoleInfo's + /// implementation needs the ActorSystem, which does not exist yet at + /// ValidateOnStart time. Same shape as and + /// . + /// + /// The application configuration, read for Cluster:Roles. + public MeshTransportOptionsValidator(IConfiguration configuration) + { + _configuration = configuration; + } + /// public ValidateOptionsResult Validate(string? name, MeshTransportOptions options) { @@ -25,13 +42,14 @@ public sealed class MeshTransportOptionsValidator : IValidateOptions()); + var isDriver = Array.IndexOf(roles, RoleParser.Driver) >= 0; + + if (isGrpc && isDriver) + { + if (string.IsNullOrWhiteSpace(options.GrpcPsk)) + { + errors.Add( + $"{MeshTransportOptions.SectionName}:{nameof(MeshTransportOptions.GrpcPsk)} is " + + $"empty. Under {MeshTransportOptions.ModeGrpc} this per-cluster pre-shared key is " + + "the whole authentication boundary of the command plane in both directions — " + + "without it this node's inbound command service fails closed and every ack it " + + "posts to central is rejected. Supply it via the environment " + + $"({MeshTransportOptions.SectionName}__{nameof(MeshTransportOptions.GrpcPsk)}), in " + + "production as ${secret:OT-MESH-PSK-}."); + } + + if (options.CentralGrpcEndpoints.Length == 0) + { + errors.Add( + $"{MeshTransportOptions.SectionName}:" + + $"{nameof(MeshTransportOptions.CentralGrpcEndpoints)} is empty. Under " + + $"{MeshTransportOptions.ModeGrpc} a driver node has nowhere to post its " + + "deployment ack, and every deployment would time out naming a node that is " + + "running perfectly. Supply them via the environment " + + $"({MeshTransportOptions.SectionName}__" + + $"{nameof(MeshTransportOptions.CentralGrpcEndpoints)}__0)."); + } + + foreach (var endpoint in options.CentralGrpcEndpoints) + { + if (!endpoint.StartsWith("http://", StringComparison.OrdinalIgnoreCase) + && !endpoint.StartsWith("https://", StringComparison.OrdinalIgnoreCase)) + { + errors.Add( + $"Central gRPC endpoint '{endpoint}' must start with 'http://' (h2c) or " + + "'https://' — it is a gRPC base address (scheme://host:port), not a bare " + + "host:port and not an Akka remoting address."); + } + } + + // Both deadlines are per-call and there is no retry behind them, so a non-positive value + // is not "wait forever" — it is a call that expires before it is made, i.e. a command + // plane that silently delivers nothing. + if (options.DispatchTimeoutSeconds <= 0) + { + errors.Add( + $"{MeshTransportOptions.SectionName}:" + + $"{nameof(MeshTransportOptions.DispatchTimeoutSeconds)} must be greater than 0 " + + $"(was {options.DispatchTimeoutSeconds})."); + } + + if (options.AckTimeoutSeconds <= 0) + { + errors.Add( + $"{MeshTransportOptions.SectionName}:" + + $"{nameof(MeshTransportOptions.AckTimeoutSeconds)} must be greater than 0 " + + $"(was {options.AckTimeoutSeconds})."); + } + } + return errors.Count == 0 ? ValidateOptionsResult.Success : ValidateOptionsResult.Fail(string.Join(" ", errors)); diff --git a/src/Core/ZB.MOM.WW.OtOpcUa.Cluster/SplitTopologyTransportValidator.cs b/src/Core/ZB.MOM.WW.OtOpcUa.Cluster/SplitTopologyTransportValidator.cs index e9a80a95..5a5bb597 100644 --- a/src/Core/ZB.MOM.WW.OtOpcUa.Cluster/SplitTopologyTransportValidator.cs +++ b/src/Core/ZB.MOM.WW.OtOpcUa.Cluster/SplitTopologyTransportValidator.cs @@ -98,14 +98,22 @@ public sealed class SplitTopologyTransportValidator : IValidateOptions /// /// The fallback keeps ~12 existing tests (and any node wired before per-cluster mesh Phase 2) - /// on exactly the pre-Phase-2 behaviour. It is not a permanent seam: Phase 6 deletes it along - /// with the DistributedPubSub branch and the single mesh. + /// on exactly the pre-Phase-2 behaviour. The DistributedPubSub branch behind it is + /// retained, not scheduled for deletion — an earlier note here promised Phase 6 would + /// remove it, which was deliberately reversed (docs/Configuration.md § "The compiled + /// transport-mode defaults were deliberately NOT flipped"): DPS is still the correct and + /// default transport for a single-mesh topology. What is being replaced is the ClusterClient + /// branch, by gRPC. /// /// The legacy DistributedPubSub topic. /// The command. diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.ControlPlane/Coordinators/ConfigPublishCoordinator.cs b/src/Server/ZB.MOM.WW.OtOpcUa.ControlPlane/Coordinators/ConfigPublishCoordinator.cs index 86787656..f71c4c74 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.ControlPlane/Coordinators/ConfigPublishCoordinator.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.ControlPlane/Coordinators/ConfigPublishCoordinator.cs @@ -68,8 +68,11 @@ public sealed class ConfigPublishCoordinator : ReceiveActor, IWithTimers /// /// /// Null in tests and on any node wired before Phase 2, where the fallback reproduces the - /// pre-Phase-2 behaviour exactly. The fallback is not a permanent seam — Phase 6 deletes it - /// along with the DistributedPubSub branch and the single mesh. + /// pre-Phase-2 behaviour exactly. The DistributedPubSub branch behind it is retained, + /// not scheduled for deletion — an earlier note here promised Phase 6 would remove it, which + /// was deliberately reversed (docs/Configuration.md § "The compiled transport-mode + /// defaults were deliberately NOT flipped"): DPS is still the correct and default transport + /// for a single-mesh topology. What is being replaced is the ClusterClient branch, by gRPC. /// private readonly IActorRef? _meshRouter; private readonly ILoggingAdapter _log = Context.GetLogger(); diff --git a/tests/Core/ZB.MOM.WW.OtOpcUa.Cluster.Tests/MeshTransportOptionsValidatorTests.cs b/tests/Core/ZB.MOM.WW.OtOpcUa.Cluster.Tests/MeshTransportOptionsValidatorTests.cs index af627a9b..1fb4ce46 100644 --- a/tests/Core/ZB.MOM.WW.OtOpcUa.Cluster.Tests/MeshTransportOptionsValidatorTests.cs +++ b/tests/Core/ZB.MOM.WW.OtOpcUa.Cluster.Tests/MeshTransportOptionsValidatorTests.cs @@ -1,3 +1,4 @@ +using Microsoft.Extensions.Configuration; using Microsoft.Extensions.Options; using Shouldly; using Xunit; @@ -10,10 +11,35 @@ namespace ZB.MOM.WW.OtOpcUa.Cluster.Tests; /// trace. Absences are the hardest class of fault to attribute in the field, so the whole point /// of these tests is that the host refuses to start instead. /// +/// +/// Role-dependent facts run in because the validator +/// resolves roles through RoleParser.ResolveEffectiveRoles, which reads +/// OTOPCUA_ROLES when Cluster:Roles is empty — a process-wide env var, so the +/// whole class is serialized against the other role-reading suites. +/// +[Collection(RoleEnvironmentCollection.Name)] public class MeshTransportOptionsValidatorTests { - private static ValidateOptionsResult Validate(MeshTransportOptions o) => - new MeshTransportOptionsValidator().Validate(MeshTransportOptions.SectionName, o); + private static ValidateOptionsResult Validate(MeshTransportOptions o, params string[] roles) + { + var pairs = new Dictionary(); + for (var i = 0; i < roles.Length; i++) + { + pairs[$"Cluster:Roles:{i}"] = roles[i]; + } + + var configuration = new ConfigurationBuilder().AddInMemoryCollection(pairs).Build(); + + return new MeshTransportOptionsValidator(configuration) + .Validate(MeshTransportOptions.SectionName, o); + } + + private static MeshTransportOptions WellFormedGrpc() => new() + { + Mode = MeshTransportOptions.ModeGrpc, + GrpcPsk = "not-a-real-key", + CentralGrpcEndpoints = ["http://central-1:4055", "https://central-2:4055"], + }; [Fact] public void Default_options_are_valid() @@ -26,10 +52,23 @@ public class MeshTransportOptionsValidatorTests [Fact] public void Unknown_mode_fails() { - var result = Validate(new MeshTransportOptions { Mode = "grpc" }); + var result = Validate(new MeshTransportOptions { Mode = "receptionist" }); result.Failed.ShouldBeTrue(); - result.FailureMessage.ShouldContain("grpc"); + result.FailureMessage.ShouldContain("receptionist"); + } + + [Fact] + public void Unknown_mode_message_names_all_three_accepted_values() + { + // An operator who mistypes the mode gets the full menu, not a partial one — Grpc became a + // legal value in Phase 0 and an error that omitted it would send them back to ClusterClient. + var result = Validate(new MeshTransportOptions { Mode = "receptionist" }); + + result.Failed.ShouldBeTrue(); + result.FailureMessage.ShouldContain(MeshTransportOptions.ModeDps); + result.FailureMessage.ShouldContain(MeshTransportOptions.ModeClusterClient); + result.FailureMessage.ShouldContain(MeshTransportOptions.ModeGrpc); } [Fact] @@ -135,4 +174,210 @@ public class MeshTransportOptionsValidatorTests ], }).Succeeded.ShouldBeTrue(); } + + // ---- Grpc mode (ClusterClient→gRPC migration, Phase 0) -------------------------------------- + + [Fact] + public void Grpc_is_an_accepted_mode() + { + Validate(WellFormedGrpc(), "driver", "cluster-SITE-A").Succeeded.ShouldBeTrue(); + } + + [Fact] + public void Grpc_mode_matching_is_case_insensitive() + { + var options = WellFormedGrpc(); + options.Mode = "grpc"; + + Validate(options, "driver", "cluster-SITE-A").Succeeded.ShouldBeTrue(); + } + + [Fact] + public void Grpc_driver_node_without_a_psk_fails() + { + var options = WellFormedGrpc(); + options.GrpcPsk = null; + + var result = Validate(options, "driver", "cluster-SITE-A"); + + result.Failed.ShouldBeTrue(); + result.FailureMessage.ShouldContain("MeshTransport:GrpcPsk"); + result.FailureMessage.ShouldContain("MeshTransport__GrpcPsk"); + } + + [Fact] + public void Grpc_driver_node_with_an_empty_psk_fails() + { + // Empty is the shape an unset environment variable actually produces, and it is exactly as + // fatal as null — a blank bearer authenticates nothing. + var options = WellFormedGrpc(); + options.GrpcPsk = " "; + + Validate(options, "driver", "cluster-SITE-A").Failed.ShouldBeTrue(); + } + + [Fact] + public void The_psk_value_is_never_echoed_in_a_failure_message() + { + // The whole point of a validator that reads a secret is that its diagnostics are safe to + // paste into a ticket. Naming the key is the contract; printing it is a leak. + var options = WellFormedGrpc(); + options.GrpcPsk = "super-secret-mesh-psk"; + options.CentralGrpcEndpoints = ["central-1:4055"]; + + var result = Validate(options, "driver", "cluster-SITE-A"); + + result.Failed.ShouldBeTrue(); + result.FailureMessage.ShouldNotContain("super-secret-mesh-psk"); + } + + [Fact] + public void Grpc_driver_node_without_central_endpoints_fails() + { + var options = WellFormedGrpc(); + options.CentralGrpcEndpoints = []; + + var result = Validate(options, "driver", "cluster-SITE-A"); + + result.Failed.ShouldBeTrue(); + result.FailureMessage.ShouldContain("MeshTransport:CentralGrpcEndpoints"); + } + + [Fact] + public void A_central_endpoint_that_is_not_an_http_url_fails() + { + var options = WellFormedGrpc(); + options.CentralGrpcEndpoints = ["central-1:4055"]; + + var result = Validate(options, "driver", "cluster-SITE-A"); + + result.Failed.ShouldBeTrue(); + result.FailureMessage.ShouldContain("central-1:4055"); + result.FailureMessage.ShouldContain("http://"); + } + + [Fact] + public void One_bad_central_endpoint_among_good_ones_still_fails() + { + var options = WellFormedGrpc(); + options.CentralGrpcEndpoints = ["http://central-1:4055", "akka.tcp://otopcua@central-2:4053"]; + + var result = Validate(options, "driver", "cluster-SITE-A"); + + result.Failed.ShouldBeTrue(); + result.FailureMessage.ShouldContain("akka.tcp://otopcua@central-2:4053"); + } + + [Fact] + public void Grpc_driver_node_with_a_non_positive_dispatch_timeout_fails() + { + var options = WellFormedGrpc(); + options.DispatchTimeoutSeconds = 0; + + var result = Validate(options, "driver", "cluster-SITE-A"); + + result.Failed.ShouldBeTrue(); + result.FailureMessage.ShouldContain("MeshTransport:DispatchTimeoutSeconds"); + } + + [Fact] + public void Grpc_driver_node_with_a_non_positive_ack_timeout_fails() + { + var options = WellFormedGrpc(); + options.AckTimeoutSeconds = -1; + + var result = Validate(options, "driver", "cluster-SITE-A"); + + result.Failed.ShouldBeTrue(); + result.FailureMessage.ShouldContain("MeshTransport:AckTimeoutSeconds"); + } + + [Fact] + public void Grpc_admin_only_node_needs_none_of_the_node_side_keys() + { + // The node-side surface is what a DRIVER uses to dial central and post its ack. An admin-only + // central node dials nodes instead and resolves per-cluster keys at runtime, so requiring + // these here would make an admin node carry a PSK and an endpoint list it never reads. + Validate( + new MeshTransportOptions { Mode = MeshTransportOptions.ModeGrpc }, + "admin", "cluster-MAIN").Succeeded.ShouldBeTrue(); + } + + [Fact] + public void Grpc_role_less_node_needs_none_of_the_node_side_keys() + { + // Every in-process test host and every single-mesh node has no declared role. Requiring the + // Grpc surface of them would turn a dark-switch value into a tree-wide config obligation. + Validate(new MeshTransportOptions { Mode = MeshTransportOptions.ModeGrpc }).Succeeded.ShouldBeTrue(); + } + + [Fact] + public void Roles_come_from_OTOPCUA_ROLES_when_Cluster_Roles_is_empty() + { + // Roles-source divergence (Gitea #532): a node configured only through the env var is still a + // driver node in Akka. A validator reading Cluster:Roles alone would exempt the exact node + // that has to dial central, and its acks would be rejected unauthenticated at runtime. + using (new EnvVarScope("OTOPCUA_ROLES", "driver,cluster-SITE-A")) + { + var options = WellFormedGrpc(); + options.GrpcPsk = null; + + Validate(options).Failed.ShouldBeTrue(); + } + } + + [Fact] + public void Dps_mode_requires_none_of_the_new_Grpc_keys() + { + // The no-new-obligations invariant: adding a third mode must not make any existing + // deployment's configuration invalid. A Dps driver node that sets nothing new still passes. + Validate( + new MeshTransportOptions { Mode = MeshTransportOptions.ModeDps }, + "driver", "cluster-SITE-A").Succeeded.ShouldBeTrue(); + } + + [Fact] + public void ClusterClient_mode_requires_none_of_the_new_Grpc_keys() + { + Validate( + new MeshTransportOptions + { + Mode = MeshTransportOptions.ModeClusterClient, + CentralContactPoints = ["akka.tcp://otopcua@central-1:4053"], + }, + "driver", "cluster-SITE-A").Succeeded.ShouldBeTrue(); + } + + [Fact] + public void ClusterPsks_defaults_to_an_empty_case_insensitive_map() + { + // Production leaves the map empty and the runtime provider falls back to the Secrets store; + // the rig populates it. Case-insensitivity matters because a ClusterId reaches this lookup + // from three sources — a role suffix, a DB column and an env key — that do not agree on case. + var options = new MeshTransportOptions(); + + options.ClusterPsks.ShouldBeEmpty(); + options.ClusterPsks["MAIN"] = "rig-key"; + options.ClusterPsks.ContainsKey("main").ShouldBeTrue(); + } + + [Fact] + public void ClusterPsks_stays_case_insensitive_after_binding_configuration() + { + // Asserting the comparer on a fresh instance proves nothing about the shape that reaches + // production: the options binder could construct its own Dictionary and drop the comparer + // silently, and the symptom would be a rig lookup that misses on a ClusterId spelled the way + // the DB spells it. Bind for real and check the comparer survived. + var configuration = new ConfigurationBuilder() + .AddInMemoryCollection(new Dictionary + { + ["MeshTransport:ClusterPsks:SITE-A"] = "rig-key", + }) + .Build(); + + var options = new MeshTransportOptions(); + configuration.GetSection(MeshTransportOptions.SectionName).Bind(options); + + options.ClusterPsks["site-a"].ShouldBe("rig-key"); + } } diff --git a/tests/Core/ZB.MOM.WW.OtOpcUa.Cluster.Tests/SplitTopologyTransportValidatorTests.cs b/tests/Core/ZB.MOM.WW.OtOpcUa.Cluster.Tests/SplitTopologyTransportValidatorTests.cs index 6d5b018c..4aff26d5 100644 --- a/tests/Core/ZB.MOM.WW.OtOpcUa.Cluster.Tests/SplitTopologyTransportValidatorTests.cs +++ b/tests/Core/ZB.MOM.WW.OtOpcUa.Cluster.Tests/SplitTopologyTransportValidatorTests.cs @@ -52,9 +52,58 @@ public class SplitTopologyTransportValidatorTests result.Failed.ShouldBeTrue(); result.FailureMessage.ShouldContain("MeshTransport:Mode"); result.FailureMessage.ShouldContain(MeshTransportOptions.ModeClusterClient); + result.FailureMessage.ShouldContain(MeshTransportOptions.ModeGrpc); result.FailureMessage.ShouldContain("SITE-A"); } + [Fact] + public void Cluster_role_driver_node_on_Grpc_mesh_transport_passes() + { + // The migration window: Grpc is the transport this rule exists to arrive at, so a rig node + // flipped to it must boot. Without this relaxation the very first Grpc rollout step — setting + // the mode on a split-topology node — is a host-start failure. + Validate( + meshMode: MeshTransportOptions.ModeGrpc, + telemetryMode: TelemetryOptions.ModeGrpc, + telemetryDialMode: null, + "driver", "cluster-SITE-A").Succeeded.ShouldBeTrue(); + } + + [Fact] + public void Cluster_role_admin_node_on_Grpc_mesh_transport_passes() + { + Validate( + meshMode: MeshTransportOptions.ModeGrpc, + telemetryMode: null, + telemetryDialMode: TelemetryDialOptions.ModeGrpc, + "admin", "cluster-SITE-A").Succeeded.ShouldBeTrue(); + } + + [Fact] + public void Cluster_role_node_on_Grpc_mesh_transport_is_matched_case_insensitively() + { + Validate( + meshMode: "grpc", + telemetryMode: TelemetryOptions.ModeGrpc, + telemetryDialMode: TelemetryDialOptions.ModeGrpc, + "admin", "driver", "cluster-MAIN").Succeeded.ShouldBeTrue(); + } + + [Fact] + public void Cluster_role_node_on_an_unrecognised_mesh_transport_still_fails() + { + // Widening the rule from one accepted value to two must not widen it to "anything that is not + // Dps" — a typo'd mode is still a mode that delivers nothing across the boundary. + var result = Validate( + meshMode: "receptionist", + telemetryMode: TelemetryOptions.ModeGrpc, + telemetryDialMode: null, + "driver", "cluster-SITE-A"); + + result.Failed.ShouldBeTrue(); + result.FailureMessage.ShouldContain("MeshTransport:Mode"); + } + [Fact] public void Cluster_role_driver_node_fully_configured_passes() { From 09db9e47de6c67c71f0c281c7143b4841ff2c7e9 Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Thu, 6 Aug 2026 03:12:09 -0400 Subject: [PATCH 03/12] =?UTF-8?q?chore(mesh):=20Batch=20A=20review=20close?= =?UTF-8?q?-out=20=E2=80=94=20fusion-invariant=20note=20on=20DispatchTimeo?= =?UTF-8?q?utSeconds,=20tracking=20updates?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- docs/plans/2026-08-06-clusterclient-to-grpc-tracking.md | 4 +++- docs/plans/2026-08-06-clusterclient-to-grpc.md.tasks.json | 6 +++--- src/Core/ZB.MOM.WW.OtOpcUa.Cluster/MeshTransportOptions.cs | 3 +++ 3 files changed, 9 insertions(+), 4 deletions(-) diff --git a/docs/plans/2026-08-06-clusterclient-to-grpc-tracking.md b/docs/plans/2026-08-06-clusterclient-to-grpc-tracking.md index a9739c34..5bc95f15 100644 --- a/docs/plans/2026-08-06-clusterclient-to-grpc-tracking.md +++ b/docs/plans/2026-08-06-clusterclient-to-grpc-tracking.md @@ -22,7 +22,7 @@ posture). | Phase | Scope | Branch | Status | PR / merge SHA | Live gate | |---|---|---|---|---|---| -| 0 | Proto + mappers + goldens; both auth interceptors; `IMeshPskProvider`; options/validators (+`Grpc` value, SplitTopology relaxation); rig dark config; stale-comment fixes; FleetStatus issue filed | `feat/grpc-phase0-contract-auth` | **NOT STARTED** | — | PENDING | +| 0 | Proto + mappers + goldens; both auth interceptors; `IMeshPskProvider`; options/validators (+`Grpc` value, SplitTopology relaxation); rig dark config; stale-comment fixes; FleetStatus issue filed | `feat/grpc-phase0-contract-auth` | **IN PROGRESS** — Batch A (0.1 + 0.3) done @ `aa21f61f`, spec ✅ + code review ✅ (0 critical/important); 0.2 next | — | PENDING | | 1A | Central→node command plane (`NodeCommandService` on the Telemetry listener; `MeshGrpcCommandDialer`; third mode branch in `CentralCommunicationActor`) — **worktree A** | `feat/grpc-node-command` | NOT STARTED | — | PENDING | | 1B | Node→central ack plane (`CentralCommandService` on the ConfigServe listener; `GrpcApplyAckSender`; third branch in `NodeCommunicationActor`) — **worktree B**, rebases on 1A | `feat/grpc-central-ack` | NOT STARTED | — | PENDING | | 2 | Baseline reproduction on pre-flip build, then full-rig cutover (`OTOPCUA_MESH_MODE` default → `Grpc`) + soak + kill drills | (rig commit) | NOT STARTED | — | PENDING | @@ -45,6 +45,8 @@ phase DoD — PR + merge are user-owned.** | 2026-08-06 | `Dps` mode survives for single-mesh topologies; only `ClusterClient` is deleted; the three stale "Phase 6 deletes DPS" comments get corrected, not obeyed | design §3.6/§4.2 | | 2026-08-06 | No transport interface — third mode branch inside the two choke-point actors (ScadaBridge §7.1 economy) | design §4.3 | | 2026-08-06 | Coexistence granularity: central's mode governs commands, each node's mode governs its acks (per-pair flips possible — better than ScadaBridge's central-wide flag) | design §3.6 | +| 2026-08-06 | `AlarmCommandMessage.operation` is a proto `string`, NOT an enum — the CLR `AlarmCommand.Operation` is an unconstrained string, and an enum would silently decode unknown operations to `UNSPECIFIED` (the drop class this migration removes). Plan T0.1's "AlarmCommand verb" enum wording was a plan defect; only `ApplyOutcome` carries the enum-0 rule. T0.2's goldens must pin pass-through of an unrecognized operation string | proto comment + this log (T0.1 review) | +| 2026-08-06 | `correlation_id` lives on the envelope AND as inner fields on the three commands whose CLR record carries one — each command round-trips independently in the T0.2 mapper; the envelope covers `AlarmCommand` (no CLR correlation member). For the three, the inner field is authoritative on decode | proto comment + this log (T0.1 review) | ## Defect log diff --git a/docs/plans/2026-08-06-clusterclient-to-grpc.md.tasks.json b/docs/plans/2026-08-06-clusterclient-to-grpc.md.tasks.json index 46cd683c..2ffb2b35 100644 --- a/docs/plans/2026-08-06-clusterclient-to-grpc.md.tasks.json +++ b/docs/plans/2026-08-06-clusterclient-to-grpc.md.tasks.json @@ -3,9 +3,9 @@ "executionModel": "opus-subagents; sequential within a worktree, 1A and 1B concurrent in separate worktrees (git worktree add ../OtOpcUa-1B feat/grpc-central-ack); 1A merges first, 1B rebases; stop at each phase DoD (PR/merge user-owned)", "worktrees": {}, "tasks": [ - {"id": 0, "subject": "Task 0.1: mesh_transport.proto + contract lock", "phase": "0", "branch": "feat/grpc-phase0-contract-auth", "classification": "standard", "parallelizableWith": ["0.3 (batch, same agent)"], "status": "pending"}, - {"id": 1, "subject": "Task 0.2: DTO mapper + round-trip goldens", "phase": "0", "branch": "feat/grpc-phase0-contract-auth", "classification": "high-risk", "parallelizableWith": [], "status": "pending", "blockedBy": [0]}, - {"id": 2, "subject": "Task 0.3: MeshTransportOptions + validators + SplitTopology relaxation", "phase": "0", "branch": "feat/grpc-phase0-contract-auth", "classification": "standard", "parallelizableWith": ["0.1 (batch, same agent)"], "status": "pending"}, + {"id": 0, "subject": "Task 0.1: mesh_transport.proto + contract lock", "phase": "0", "branch": "feat/grpc-phase0-contract-auth", "classification": "standard", "parallelizableWith": ["0.3 (batch, same agent)"], "status": "completed", "commit": "c0ccfa27", "review": "spec ✅ + code ✅"}, + {"id": 1, "subject": "Task 0.2: DTO mapper + round-trip goldens", "phase": "0", "branch": "feat/grpc-phase0-contract-auth", "classification": "high-risk", "parallelizableWith": [], "status": "in_progress", "blockedBy": [0]}, + {"id": 2, "subject": "Task 0.3: MeshTransportOptions + validators + SplitTopology relaxation", "phase": "0", "branch": "feat/grpc-phase0-contract-auth", "classification": "standard", "parallelizableWith": ["0.1 (batch, same agent)"], "status": "completed", "commit": "aa21f61f", "review": "spec ✅ + code ✅"}, {"id": 3, "subject": "Task 0.4: node-side MeshCommandAuthInterceptor", "phase": "0", "branch": "feat/grpc-phase0-contract-auth", "classification": "high-risk", "parallelizableWith": ["0.5 (batch, same agent)"], "status": "pending", "blockedBy": [2]}, {"id": 4, "subject": "Task 0.5: IMeshPskProvider + CentralCommandAuthInterceptor", "phase": "0", "branch": "feat/grpc-phase0-contract-auth", "classification": "high-risk", "parallelizableWith": ["0.4 (batch, same agent)"], "status": "pending", "blockedBy": [2]}, {"id": 5, "subject": "Task 0.6: rig config + FleetStatus issue + docs stub + Phase 0 gate", "phase": "0", "branch": "feat/grpc-phase0-contract-auth", "classification": "small", "parallelizableWith": [], "status": "pending", "blockedBy": [0, 1, 2, 3, 4]}, diff --git a/src/Core/ZB.MOM.WW.OtOpcUa.Cluster/MeshTransportOptions.cs b/src/Core/ZB.MOM.WW.OtOpcUa.Cluster/MeshTransportOptions.cs index d3e41863..a52c515d 100644 --- a/src/Core/ZB.MOM.WW.OtOpcUa.Cluster/MeshTransportOptions.cs +++ b/src/Core/ZB.MOM.WW.OtOpcUa.Cluster/MeshTransportOptions.cs @@ -131,6 +131,9 @@ public sealed class MeshTransportOptions /// unbounded call to one unreachable node would stall the commands queued behind it. Expiry /// is logged and dropped — the transport is fire-and-forget and layers no retry, so the /// deploy path's 2-minute apply deadline stays the only recovery mechanism. + /// Validation of this key is scoped to driver-role nodes; that covers the central pair only + /// because central nodes are always fused (admin,driver,cluster-*). A pure-admin + /// central node would leave this unvalidated on the node that uses it. /// public int DispatchTimeoutSeconds { get; set; } = 10; From 545413a83534e60bb7a10128063dfb9a3432d148 Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Thu, 6 Aug 2026 03:19:11 -0400 Subject: [PATCH 04/12] feat(mesh): DTO mapper with round-trip goldens --- .../Messages/Mesh/MeshTransportDtoMapper.cs | 391 +++++++++++++++ .../MeshTransportDtoMapperTests.cs | 451 ++++++++++++++++++ 2 files changed, 842 insertions(+) create mode 100644 src/Core/ZB.MOM.WW.OtOpcUa.Commons/Messages/Mesh/MeshTransportDtoMapper.cs create mode 100644 tests/Core/ZB.MOM.WW.OtOpcUa.Commons.Tests/MeshTransportDtoMapperTests.cs diff --git a/src/Core/ZB.MOM.WW.OtOpcUa.Commons/Messages/Mesh/MeshTransportDtoMapper.cs b/src/Core/ZB.MOM.WW.OtOpcUa.Commons/Messages/Mesh/MeshTransportDtoMapper.cs new file mode 100644 index 00000000..1b83ef7d --- /dev/null +++ b/src/Core/ZB.MOM.WW.OtOpcUa.Commons/Messages/Mesh/MeshTransportDtoMapper.cs @@ -0,0 +1,391 @@ +using Google.Protobuf.WellKnownTypes; +using ZB.MOM.WW.OtOpcUa.Commons.Messages.Admin; +using ZB.MOM.WW.OtOpcUa.Commons.Messages.Deploy; +using ZB.MOM.WW.OtOpcUa.Commons.OpcUa; +using ZB.MOM.WW.OtOpcUa.Commons.Protos.MeshTransport.V1; +using ZB.MOM.WW.OtOpcUa.Commons.Types; + +namespace ZB.MOM.WW.OtOpcUa.Commons.Messages.Mesh; + +/// +/// The one translation between the five cross-mesh command records and their +/// mesh_transport.v1 wire shapes (ClusterClient→gRPC migration Phase 0). Both directions live +/// here so a field can only be added to one side without the other by making this file +/// visibly asymmetric. +/// +/// +/// +/// Nothing here guesses. The transport it replaces — an Akka ClusterClient receptionist — +/// dropped an undeliverable command silently, and the entire point of the migration is to stop +/// doing that. So every decode that cannot produce a faithful record throws: an unset oneof, an +/// APPLY_OUTCOME_UNSPECIFIED ack, an enum value from a newer peer. A loud failure on one +/// command is recoverable; a deployment reported applied because an absent enum field decoded to +/// 0 is not. +/// +/// +/// Presence, not emptiness. The two nullable members that cross this contract — +/// and — are +/// optional string on the wire, so null maps to an absent field and "" to a +/// present-but-empty one. Collapsing them would report an operator's deliberately blank comment +/// as "no comment supplied", and a blank failure reason as "no reason recorded". +/// uses a message field, whose own absence carries the +/// null. +/// +/// +/// Correlation is written twice and read once. Encode stamps the envelope's +/// correlation_id and the command's own from the same source. Decode takes the command's +/// own for the three records that declare one — that member is domain state the record must +/// carry back verbatim — and leaves the envelope's to the transport, which is the only place +/// 's correlation can live at all since the record has no member for it. +/// +/// +public static class MeshTransportDtoMapper +{ + // ---------------------------------------------------------------- DispatchDeployment + + /// Projects onto its wire form. + /// The domain dispatch command. + /// The wire . + public static DispatchDeploymentCommand ToProto(DispatchDeployment message) + { + ArgumentNullException.ThrowIfNull(message); + + return new DispatchDeploymentCommand + { + DeploymentId = message.DeploymentId.ToString(), + RevisionHash = message.RevisionHash.Value ?? "", + CorrelationId = message.CorrelationId.ToString(), + }; + } + + /// Rebuilds the domain record from its wire form. + /// The wire dispatch command. + /// The domain . + /// An id field is not a hyphenless GUID. + /// The revision hash is absent or blank. + public static DispatchDeployment FromProto(DispatchDeploymentCommand proto) + { + ArgumentNullException.ThrowIfNull(proto); + + return new DispatchDeployment( + DeploymentId.Parse(proto.DeploymentId), + RevisionHash.Parse(proto.RevisionHash), + CorrelationId.Parse(proto.CorrelationId)); + } + + // ---------------------------------------------------------------- RestartDriver + + /// Projects onto its wire form. + /// The domain restart command. + /// The wire . + public static RestartDriverCommand ToProto(RestartDriver message) + { + ArgumentNullException.ThrowIfNull(message); + + // Required (non-optional) proto string setters throw on null and the record's nullable-ref + // annotations are compile-time only, so a null reaching here is already a producer bug. Coalesce + // rather than throw out of the mapper: a command with a blank field is diagnosable downstream, + // whereas a throw here is the silent drop this transport exists to remove. + return new RestartDriverCommand + { + ClusterId = message.ClusterId ?? "", + DriverInstanceId = message.DriverInstanceId ?? "", + ActorByUserName = message.ActorByUserName ?? "", + CorrelationId = message.CorrelationId.ToString("N"), + }; + } + + /// Rebuilds the domain record from its wire form. + /// The wire restart command. + /// The domain . + /// The correlation id is not a hyphenless GUID. + public static RestartDriver FromProto(RestartDriverCommand proto) + { + ArgumentNullException.ThrowIfNull(proto); + + return new RestartDriver( + proto.ClusterId, + proto.DriverInstanceId, + proto.ActorByUserName, + Guid.ParseExact(proto.CorrelationId, "N")); + } + + // ---------------------------------------------------------------- ReconnectDriver + + /// Projects onto its wire form. + /// The domain reconnect command. + /// The wire . + public static ReconnectDriverCommand ToProto(ReconnectDriver message) + { + ArgumentNullException.ThrowIfNull(message); + + return new ReconnectDriverCommand + { + ClusterId = message.ClusterId ?? "", + DriverInstanceId = message.DriverInstanceId ?? "", + ActorByUserName = message.ActorByUserName ?? "", + CorrelationId = message.CorrelationId.ToString("N"), + }; + } + + /// Rebuilds the domain record from its wire form. + /// The wire reconnect command. + /// The domain . + /// The correlation id is not a hyphenless GUID. + public static ReconnectDriver FromProto(ReconnectDriverCommand proto) + { + ArgumentNullException.ThrowIfNull(proto); + + return new ReconnectDriver( + proto.ClusterId, + proto.DriverInstanceId, + proto.ActorByUserName, + Guid.ParseExact(proto.CorrelationId, "N")); + } + + // ---------------------------------------------------------------- AlarmCommand + + /// Projects onto its wire form. + /// The domain alarm command. + /// The wire . + /// + /// is copied verbatim. It is an unconstrained string in the + /// record and stays one on the wire, so an operation this build does not recognise reaches the + /// far side intact instead of collapsing onto an enum's UNSPECIFIED and being dropped. + /// + public static AlarmCommandMessage ToProto(AlarmCommand message) + { + ArgumentNullException.ThrowIfNull(message); + + var proto = new AlarmCommandMessage + { + AlarmId = message.AlarmId ?? "", + Operation = message.Operation ?? "", + User = message.User ?? "", + }; + + if (message.Comment is not null) + proto.Comment = message.Comment; + if (message.UnshelveAtUtc is { } unshelveAt) + proto.UnshelveAtUtc = ToUtcTimestamp(unshelveAt); + + return proto; + } + + /// Rebuilds the domain record from its wire form. + /// The wire alarm command. + /// The domain . + public static AlarmCommand FromProto(AlarmCommandMessage proto) + { + ArgumentNullException.ThrowIfNull(proto); + + return new AlarmCommand( + proto.AlarmId, + proto.Operation, + proto.User, + proto.HasComment ? proto.Comment : null, + proto.UnshelveAtUtc?.ToDateTime()); + } + + // ---------------------------------------------------------------- ApplyAck + + /// Projects onto its wire form. + /// The domain per-node apply acknowledgment. + /// The wire . + /// + /// is not a declared . + /// + public static ApplyAckMessage ToProto(ApplyAck message) + { + ArgumentNullException.ThrowIfNull(message); + + var proto = new ApplyAckMessage + { + DeploymentId = message.DeploymentId.ToString(), + NodeId = message.NodeId.Value ?? "", + Outcome = message.Outcome switch + { + ApplyAckOutcome.Applied => ApplyOutcome.Applied, + ApplyAckOutcome.Failed => ApplyOutcome.Failed, + _ => throw new ArgumentOutOfRangeException( + nameof(message), message.Outcome, "Unknown ApplyAckOutcome"), + }, + CorrelationId = message.CorrelationId.ToString(), + }; + + if (message.FailureReason is not null) + proto.FailureReason = message.FailureReason; + + return proto; + } + + /// Rebuilds the domain record from its wire form. + /// The wire apply acknowledgment. + /// The domain . + /// + /// The outcome is APPLY_OUTCOME_UNSPECIFIED or a value this build does not know. + /// + /// An id field is not a hyphenless GUID. + /// The node id is absent or blank. + public static ApplyAck FromProto(ApplyAckMessage proto) + { + ArgumentNullException.ThrowIfNull(proto); + + // proto3 decodes an absent enum field as 0, and the deploy path seals a revision on Applied. + // ApplyAckOutcome has no Unknown member to land on, so an unspecified or unrecognised outcome + // must refuse rather than pick one of the two real answers — either choice would be a lie about + // what the node said, and the "Applied" lie seals a deployment nobody applied. + var outcome = proto.Outcome switch + { + ApplyOutcome.Applied => ApplyAckOutcome.Applied, + ApplyOutcome.Failed => ApplyAckOutcome.Failed, + _ => throw new InvalidOperationException( + $"ApplyAckMessage carries outcome '{proto.Outcome}', which has no ApplyAckOutcome " + + "equivalent. Refusing to decode rather than report an unsaid deployment result."), + }; + + return new ApplyAck( + DeploymentId.Parse(proto.DeploymentId), + NodeId.Parse(proto.NodeId), + outcome, + proto.HasFailureReason ? proto.FailureReason : null, + CorrelationId.Parse(proto.CorrelationId)); + } + + // ---------------------------------------------------------------- Envelope + + /// Wraps a dispatch command in its transport envelope. + /// The domain dispatch command. + /// The envelope, correlated from the record's own CorrelationId. + public static CommandEnvelope ToEnvelope(DispatchDeployment message) + { + ArgumentNullException.ThrowIfNull(message); + + return new CommandEnvelope + { + CorrelationId = message.CorrelationId.ToString(), + DispatchDeployment = ToProto(message), + }; + } + + /// Wraps a restart command in its transport envelope. + /// The domain restart command. + /// The envelope, correlated from the record's own CorrelationId. + public static CommandEnvelope ToEnvelope(RestartDriver message) + { + ArgumentNullException.ThrowIfNull(message); + + return new CommandEnvelope + { + CorrelationId = message.CorrelationId.ToString("N"), + RestartDriver = ToProto(message), + }; + } + + /// Wraps a reconnect command in its transport envelope. + /// The domain reconnect command. + /// The envelope, correlated from the record's own CorrelationId. + public static CommandEnvelope ToEnvelope(ReconnectDriver message) + { + ArgumentNullException.ThrowIfNull(message); + + return new CommandEnvelope + { + CorrelationId = message.CorrelationId.ToString("N"), + ReconnectDriver = ToProto(message), + }; + } + + /// Wraps an alarm command in its transport envelope. + /// The domain alarm command. + /// + /// Transport correlation. Explicit because is the one record with no + /// correlation member of its own, so the envelope is the only place it can be carried. + /// + /// The envelope. + public static CommandEnvelope ToEnvelope(AlarmCommand message, CorrelationId correlationId) + { + ArgumentNullException.ThrowIfNull(message); + + return new CommandEnvelope + { + CorrelationId = correlationId.ToString(), + AlarmCommand = ToProto(message), + }; + } + + /// + /// Untyped dispatch for the transport router, which holds its payload as + /// (an ). + /// + /// One of the four central→node command records. + /// + /// Consulted only when is an . The + /// other three carry their own correlation and it is authoritative for them, so passing a + /// different value here cannot make the envelope disagree with the command inside it. + /// + /// The envelope produced by the matching typed overload. + /// + /// is not one of the four handled command records — the transport + /// would otherwise carry a payload the far side cannot decode. + /// + public static CommandEnvelope ToEnvelope(object message, CorrelationId alarmCorrelationId) + { + ArgumentNullException.ThrowIfNull(message); + + return message switch + { + DispatchDeployment m => ToEnvelope(m), + RestartDriver m => ToEnvelope(m), + ReconnectDriver m => ToEnvelope(m), + AlarmCommand m => ToEnvelope(m, alarmCorrelationId), + _ => throw new ArgumentOutOfRangeException( + nameof(message), message.GetType().FullName, "Not a cross-mesh command record"), + }; + } + + /// Unwraps a transport envelope into the domain record it carries. + /// The received envelope. + /// + /// One of , , + /// or , boxed — the node-side handlers + /// match on the concrete type exactly as they do under the DPS transport. + /// + /// + /// The oneof is unset or names a case this build does not handle. Kept unreachable by + /// MeshTransportProtoContract and its contract-lock test; it throws anyway because the + /// alternative is a command that crosses the mesh and disappears. + /// + public static object FromEnvelope(CommandEnvelope envelope) + { + ArgumentNullException.ThrowIfNull(envelope); + + return envelope.CommandCase switch + { + CommandEnvelope.CommandOneofCase.DispatchDeployment => FromProto(envelope.DispatchDeployment), + CommandEnvelope.CommandOneofCase.RestartDriver => FromProto(envelope.RestartDriver), + CommandEnvelope.CommandOneofCase.ReconnectDriver => FromProto(envelope.ReconnectDriver), + CommandEnvelope.CommandOneofCase.AlarmCommand => FromProto(envelope.AlarmCommand), + _ => throw new InvalidOperationException( + $"CommandEnvelope (correlation '{envelope.CorrelationId}') carries oneof case " + + $"'{envelope.CommandCase}', which this build does not handle."), + }; + } + + /// + /// Converts a domain to a proto without ever + /// throwing on Kind: Utc is used as-is, Local is converted, and Unspecified is assumed + /// already-UTC (the producer contract for every …Utc member here). + /// + private static Timestamp ToUtcTimestamp(DateTime dt) + { + var utc = dt.Kind switch + { + DateTimeKind.Utc => dt, + DateTimeKind.Local => dt.ToUniversalTime(), + _ => DateTime.SpecifyKind(dt, DateTimeKind.Utc), + }; + + return Timestamp.FromDateTime(utc); + } +} diff --git a/tests/Core/ZB.MOM.WW.OtOpcUa.Commons.Tests/MeshTransportDtoMapperTests.cs b/tests/Core/ZB.MOM.WW.OtOpcUa.Commons.Tests/MeshTransportDtoMapperTests.cs new file mode 100644 index 00000000..0177a85a --- /dev/null +++ b/tests/Core/ZB.MOM.WW.OtOpcUa.Commons.Tests/MeshTransportDtoMapperTests.cs @@ -0,0 +1,451 @@ +using Google.Protobuf; +using Shouldly; +using Xunit; +using ZB.MOM.WW.OtOpcUa.Commons.Messages.Admin; +using ZB.MOM.WW.OtOpcUa.Commons.Messages.Deploy; +using ZB.MOM.WW.OtOpcUa.Commons.Messages.Mesh; +using ZB.MOM.WW.OtOpcUa.Commons.OpcUa; +using ZB.MOM.WW.OtOpcUa.Commons.Protos; +using ZB.MOM.WW.OtOpcUa.Commons.Protos.MeshTransport.V1; +using ZB.MOM.WW.OtOpcUa.Commons.Types; + +namespace ZB.MOM.WW.OtOpcUa.Commons.Tests; + +/// +/// Round-trip goldens for — the ClusterClient→gRPC command +/// plane's only data contract. Every golden goes domain → proto → serialized bytes → proto → +/// domain, because the failure class this suite exists to catch is not a mis-assigned property but a +/// member that survives an in-memory copy and vanishes on the wire: proto3 explicit presence is a +/// serialization-level property, so a null-vs-"" or absent-vs-epoch mistake is invisible +/// unless the bytes are actually round-tripped. +/// +/// +/// Each record gets one all-members-non-default golden plus a null-variant golden for every +/// nullable member, since a nullable member is exactly where a silent drop hides — the record still +/// compares equal to a fresh instance built with the default, so nothing else in the system notices. +/// +public class MeshTransportDtoMapperTests +{ + // Deliberately non-default in every position: distinct GUIDs (so a swapped assignment cannot pass), + // a real 64-char lowercase SHA-256 hex, and strings that are neither empty nor equal to each other. + private static readonly Guid DeploymentGuid = Guid.ParseExact("11111111111111111111111111111111", "N"); + private static readonly Guid CorrelationGuid = Guid.ParseExact("22222222222222222222222222222222", "N"); + private static readonly Guid EnvelopeGuid = Guid.ParseExact("33333333333333333333333333333333", "N"); + + private const string Sha256Hex = "9f86d081884c7d659a2feaa0c55ad015a3bf4f1b2b0b822cd15d6c15b0f00a08"; + + /// Serializes and re-parses so presence is proven on the wire. + private static T Wire(T message) + where T : IMessage, new() + => new MessageParser(() => new T()).ParseFrom(message.ToByteArray()); + + // ---------------------------------------------------------------- DispatchDeployment + + [Fact] + public void DispatchDeployment_RoundTrips_WithEveryMemberSet() + { + var domain = new DispatchDeployment( + new DeploymentId(DeploymentGuid), + new RevisionHash(Sha256Hex), + new CorrelationId(CorrelationGuid)); + + MeshTransportDtoMapper.FromProto(Wire(MeshTransportDtoMapper.ToProto(domain))).ShouldBe(domain); + } + + [Fact] + public void DispatchDeployment_EncodesIdsInTheirHexNForm() + { + var proto = MeshTransportDtoMapper.ToProto(new DispatchDeployment( + new DeploymentId(DeploymentGuid), new RevisionHash(Sha256Hex), new CorrelationId(CorrelationGuid))); + + proto.DeploymentId.ShouldBe(DeploymentGuid.ToString("N")); + proto.RevisionHash.ShouldBe(Sha256Hex); + proto.CorrelationId.ShouldBe(CorrelationGuid.ToString("N")); + } + + // ---------------------------------------------------------------- RestartDriver / ReconnectDriver + + [Fact] + public void RestartDriver_RoundTrips_WithEveryMemberSet() + { + var domain = new RestartDriver("cluster-alpha", "drv-modbus-7", "jdoe@zb.local", CorrelationGuid); + + MeshTransportDtoMapper.FromProto(Wire(MeshTransportDtoMapper.ToProto(domain))).ShouldBe(domain); + } + + [Fact] + public void ReconnectDriver_RoundTrips_WithEveryMemberSet() + { + var domain = new ReconnectDriver("cluster-beta", "drv-s7-3", "aoperator@zb.local", CorrelationGuid); + + MeshTransportDtoMapper.FromProto(Wire(MeshTransportDtoMapper.ToProto(domain))).ShouldBe(domain); + } + + // ---------------------------------------------------------------- AlarmCommand + + [Fact] + public void AlarmCommand_RoundTrips_WithEveryMemberSet() + { + var domain = new AlarmCommand( + "alarm-42", + "TimedShelve", + "jdoe@zb.local", + "shelving for maintenance", + new DateTime(2026, 8, 6, 13, 45, 12, 345, DateTimeKind.Utc).AddTicks(6789)); + + MeshTransportDtoMapper.FromProto(Wire(MeshTransportDtoMapper.ToProto(domain))).ShouldBe(domain); + } + + [Fact] + public void AlarmCommand_RoundTrips_WithNullComment() + { + var domain = new AlarmCommand("alarm-42", "Acknowledge", "jdoe", null, null); + + var back = MeshTransportDtoMapper.FromProto(Wire(MeshTransportDtoMapper.ToProto(domain))); + + back.ShouldBe(domain); + back.Comment.ShouldBeNull(); + back.UnshelveAtUtc.ShouldBeNull(); + } + + [Fact] + public void AlarmCommand_EmptyComment_StaysEmpty_AndIsNotNull() + { + // The trap: proto3 without explicit presence decodes an absent string as "", collapsing + // null and "" onto one value. `optional string comment` keeps them distinct — an operator who + // acknowledged with a deliberately blank comment must not be reported as having supplied none. + var domain = new AlarmCommand("alarm-42", "AddComment", "jdoe", "", null); + + var back = MeshTransportDtoMapper.FromProto(Wire(MeshTransportDtoMapper.ToProto(domain))); + + back.ShouldBe(domain); + back.Comment.ShouldNotBeNull(); + back.Comment.ShouldBeEmpty(); + } + + [Fact] + public void AlarmCommand_UnshelveAt_RoundTripsAsUtc() + { + // DateTime equality compares ticks and ignores Kind, so record equality alone would pass on a + // Local-Kind result whose instant is hours off once anything re-interprets it. Assert Kind. + var when = new DateTime(2026, 8, 6, 13, 45, 12, DateTimeKind.Utc).AddTicks(1234567); + var domain = new AlarmCommand("alarm-42", "TimedShelve", "jdoe", null, when); + + var back = MeshTransportDtoMapper.FromProto(Wire(MeshTransportDtoMapper.ToProto(domain))); + + back.UnshelveAtUtc.ShouldNotBeNull(); + back.UnshelveAtUtc!.Value.Kind.ShouldBe(DateTimeKind.Utc); + back.UnshelveAtUtc.Value.Ticks.ShouldBe(when.Ticks); + } + + [Fact] + public void AlarmCommand_UnrecognizedOperation_RoundTripsVerbatim() + { + // Controller decision: `operation` stays a proto string precisely so an operation this build + // does not know travels intact rather than decoding to an enum's UNSPECIFIED and being dropped. + var domain = new AlarmCommand("alarm-42", "SomeFutureOperation", "jdoe", null, null); + + var back = MeshTransportDtoMapper.FromProto(Wire(MeshTransportDtoMapper.ToProto(domain))); + + back.ShouldBe(domain); + back.Operation.ShouldBe("SomeFutureOperation"); + } + + // ---------------------------------------------------------------- ApplyAck + + [Theory] + [InlineData(ApplyAckOutcome.Applied)] + [InlineData(ApplyAckOutcome.Failed)] + public void ApplyAck_RoundTrips_WithEveryMemberSet(ApplyAckOutcome outcome) + { + var domain = new ApplyAck( + new DeploymentId(DeploymentGuid), + new NodeId("site-a-1"), + outcome, + "artifact fetch failed: sha mismatch", + new CorrelationId(CorrelationGuid)); + + MeshTransportDtoMapper.FromProto(Wire(MeshTransportDtoMapper.ToProto(domain))).ShouldBe(domain); + } + + [Fact] + public void ApplyAck_RoundTrips_WithNullFailureReason() + { + var domain = new ApplyAck( + new DeploymentId(DeploymentGuid), + new NodeId("site-a-1"), + ApplyAckOutcome.Applied, + null, + new CorrelationId(CorrelationGuid)); + + var back = MeshTransportDtoMapper.FromProto(Wire(MeshTransportDtoMapper.ToProto(domain))); + + back.ShouldBe(domain); + back.FailureReason.ShouldBeNull(); + } + + [Fact] + public void ApplyAck_EmptyFailureReason_StaysEmpty_AndIsNotNull() + { + var domain = new ApplyAck( + new DeploymentId(DeploymentGuid), + new NodeId("site-a-1"), + ApplyAckOutcome.Failed, + "", + new CorrelationId(CorrelationGuid)); + + var back = MeshTransportDtoMapper.FromProto(Wire(MeshTransportDtoMapper.ToProto(domain))); + + back.ShouldBe(domain); + back.FailureReason.ShouldNotBeNull(); + back.FailureReason.ShouldBeEmpty(); + } + + [Fact] + public void ApplyAck_UnspecifiedOutcome_Throws_RatherThanDecodingToAMeaningfulValue() + { + // The single highest-stakes decode in this contract. proto3 hands back 0 for an absent enum + // field, and the deploy path seals on Applied — so 0 must never resolve to Applied (or to + // Failed, which would fail a deployment that actually succeeded). ApplyAckOutcome has no + // Unknown member to land on, so refusal is the only honest answer. + var proto = new ApplyAckMessage + { + DeploymentId = DeploymentGuid.ToString("N"), + NodeId = "site-a-1", + Outcome = ApplyOutcome.Unspecified, + CorrelationId = CorrelationGuid.ToString("N"), + }; + + var ex = Should.Throw(() => MeshTransportDtoMapper.FromProto(proto)); + ex.Message.ShouldContain(nameof(ApplyOutcome.Unspecified)); + } + + [Fact] + public void ApplyAck_UnknownOutcomeValue_Throws() + { + // A peer built against a newer contract with a third outcome. Same rule: refuse, do not guess. + var proto = new ApplyAckMessage + { + DeploymentId = DeploymentGuid.ToString("N"), + NodeId = "site-a-1", + Outcome = (ApplyOutcome)99, + CorrelationId = CorrelationGuid.ToString("N"), + }; + + Should.Throw(() => MeshTransportDtoMapper.FromProto(proto)) + .Message.ShouldContain("99"); + } + + [Fact] + public void ApplyAck_UnknownClrOutcome_ThrowsOnEncode() + { + var domain = new ApplyAck( + new DeploymentId(DeploymentGuid), + new NodeId("site-a-1"), + (ApplyAckOutcome)99, + null, + new CorrelationId(CorrelationGuid)); + + Should.Throw(() => MeshTransportDtoMapper.ToProto(domain)); + } + + // ---------------------------------------------------------------- Envelope + + [Fact] + public void Envelope_RoundTrips_DispatchDeployment() + { + var domain = new DispatchDeployment( + new DeploymentId(DeploymentGuid), new RevisionHash(Sha256Hex), new CorrelationId(CorrelationGuid)); + + var envelope = Wire(MeshTransportDtoMapper.ToEnvelope(domain)); + + envelope.CommandCase.ShouldBe(CommandEnvelope.CommandOneofCase.DispatchDeployment); + MeshTransportDtoMapper.FromEnvelope(envelope).ShouldBe(domain); + } + + [Fact] + public void Envelope_RoundTrips_RestartDriver() + { + var domain = new RestartDriver("cluster-alpha", "drv-modbus-7", "jdoe", CorrelationGuid); + + var envelope = Wire(MeshTransportDtoMapper.ToEnvelope(domain)); + + envelope.CommandCase.ShouldBe(CommandEnvelope.CommandOneofCase.RestartDriver); + MeshTransportDtoMapper.FromEnvelope(envelope).ShouldBe(domain); + } + + [Fact] + public void Envelope_RoundTrips_ReconnectDriver() + { + var domain = new ReconnectDriver("cluster-beta", "drv-s7-3", "jdoe", CorrelationGuid); + + var envelope = Wire(MeshTransportDtoMapper.ToEnvelope(domain)); + + envelope.CommandCase.ShouldBe(CommandEnvelope.CommandOneofCase.ReconnectDriver); + MeshTransportDtoMapper.FromEnvelope(envelope).ShouldBe(domain); + } + + [Fact] + public void Envelope_RoundTrips_AlarmCommand() + { + var domain = new AlarmCommand("alarm-42", "Acknowledge", "jdoe", "ack'd", null); + + var envelope = Wire(MeshTransportDtoMapper.ToEnvelope(domain, new CorrelationId(EnvelopeGuid))); + + envelope.CommandCase.ShouldBe(CommandEnvelope.CommandOneofCase.AlarmCommand); + envelope.CorrelationId.ShouldBe(EnvelopeGuid.ToString("N")); + MeshTransportDtoMapper.FromEnvelope(envelope).ShouldBe(domain); + } + + [Fact] + public void Envelope_SetsBothCorrelationIds_ForTheThreeSelfCorrelatingCommands() + { + // Encode writes envelope and inner correlation from the record's own member, so the two can + // only diverge if a peer produced them apart — see the decode-precedence golden below. + var dispatch = MeshTransportDtoMapper.ToEnvelope(new DispatchDeployment( + new DeploymentId(DeploymentGuid), new RevisionHash(Sha256Hex), new CorrelationId(CorrelationGuid))); + dispatch.CorrelationId.ShouldBe(CorrelationGuid.ToString("N")); + dispatch.DispatchDeployment.CorrelationId.ShouldBe(CorrelationGuid.ToString("N")); + + var restart = MeshTransportDtoMapper.ToEnvelope( + new RestartDriver("c", "d", "u", CorrelationGuid)); + restart.CorrelationId.ShouldBe(CorrelationGuid.ToString("N")); + restart.RestartDriver.CorrelationId.ShouldBe(CorrelationGuid.ToString("N")); + + var reconnect = MeshTransportDtoMapper.ToEnvelope( + new ReconnectDriver("c", "d", "u", CorrelationGuid)); + reconnect.CorrelationId.ShouldBe(CorrelationGuid.ToString("N")); + reconnect.ReconnectDriver.CorrelationId.ShouldBe(CorrelationGuid.ToString("N")); + } + + [Fact] + public void Envelope_InnerCorrelationWins_OnDecode() + { + // Controller decision: the inner field is the domain member, the envelope's is transport + // metadata. If a peer ever writes them apart, the record must carry what the record declared. + var envelope = Wire(new CommandEnvelope + { + CorrelationId = EnvelopeGuid.ToString("N"), + DispatchDeployment = new DispatchDeploymentCommand + { + DeploymentId = DeploymentGuid.ToString("N"), + RevisionHash = Sha256Hex, + CorrelationId = CorrelationGuid.ToString("N"), + }, + }); + + var decoded = MeshTransportDtoMapper.FromEnvelope(envelope).ShouldBeOfType(); + decoded.CorrelationId.Value.ShouldBe(CorrelationGuid); + decoded.CorrelationId.Value.ShouldNotBe(EnvelopeGuid); + } + + [Fact] + public void Envelope_InnerCorrelationWins_OnDecode_ForDriverCommands() + { + var restart = Wire(new CommandEnvelope + { + CorrelationId = EnvelopeGuid.ToString("N"), + RestartDriver = new RestartDriverCommand + { + ClusterId = "c", + DriverInstanceId = "d", + ActorByUserName = "u", + CorrelationId = CorrelationGuid.ToString("N"), + }, + }); + MeshTransportDtoMapper.FromEnvelope(restart).ShouldBeOfType() + .CorrelationId.ShouldBe(CorrelationGuid); + + var reconnect = Wire(new CommandEnvelope + { + CorrelationId = EnvelopeGuid.ToString("N"), + ReconnectDriver = new ReconnectDriverCommand + { + ClusterId = "c", + DriverInstanceId = "d", + ActorByUserName = "u", + CorrelationId = CorrelationGuid.ToString("N"), + }, + }); + MeshTransportDtoMapper.FromEnvelope(reconnect).ShouldBeOfType() + .CorrelationId.ShouldBe(CorrelationGuid); + } + + [Fact] + public void FromEnvelope_UnsetOneof_Throws() + { + // Unreachable while MeshTransportProtoContractTests holds the oneof locked, which is exactly + // why it must throw rather than return null: the alternative is a command that crosses the + // mesh, decodes to nothing and is dropped without a trace. + var ex = Should.Throw( + () => MeshTransportDtoMapper.FromEnvelope(new CommandEnvelope { CorrelationId = "x" })); + + ex.Message.ShouldContain(nameof(CommandEnvelope.CommandOneofCase.None)); + } + + [Fact] + public void ToEnvelope_UnknownMessageType_Throws() + { + Should.Throw( + () => MeshTransportDtoMapper.ToEnvelope("not a command", new CorrelationId(EnvelopeGuid))); + } + + [Fact] + public void EveryHandledOneofCase_DecodesToADomainRecord() + { + // Ties the mapper to the contract-lock list: a fifth oneof case added to the proto and to + // HandledCases without a mapper arm fails here rather than at runtime on the far side. + var samples = new Dictionary + { + [CommandEnvelope.CommandOneofCase.DispatchDeployment] = MeshTransportDtoMapper.ToEnvelope( + new DispatchDeployment(new DeploymentId(DeploymentGuid), new RevisionHash(Sha256Hex), + new CorrelationId(CorrelationGuid))), + [CommandEnvelope.CommandOneofCase.RestartDriver] = MeshTransportDtoMapper.ToEnvelope( + new RestartDriver("c", "d", "u", CorrelationGuid)), + [CommandEnvelope.CommandOneofCase.ReconnectDriver] = MeshTransportDtoMapper.ToEnvelope( + new ReconnectDriver("c", "d", "u", CorrelationGuid)), + [CommandEnvelope.CommandOneofCase.AlarmCommand] = MeshTransportDtoMapper.ToEnvelope( + new AlarmCommand("a", "Acknowledge", "u", null, null), new CorrelationId(EnvelopeGuid)), + }; + + samples.Keys.ShouldBe(MeshTransportProtoContract.HandledCases, ignoreOrder: true); + + foreach (var (expectedCase, envelope) in samples) + { + envelope.CommandCase.ShouldBe(expectedCase); + MeshTransportDtoMapper.FromEnvelope(Wire(envelope)).ShouldNotBeNull(); + } + } + + // ---------------------------------------------------------------- ToEnvelope(object …) dispatch + + [Fact] + public void ToEnvelope_ObjectDispatch_MatchesTheTypedOverloads() + { + // The router holds its command as MeshCommand.Message (an object), so the untyped entry point + // must produce byte-identical envelopes to the typed ones it delegates to. + var alarmCorrelation = new CorrelationId(EnvelopeGuid); + + object[] commands = + [ + new DispatchDeployment(new DeploymentId(DeploymentGuid), new RevisionHash(Sha256Hex), + new CorrelationId(CorrelationGuid)), + new RestartDriver("c", "d", "u", CorrelationGuid), + new ReconnectDriver("c", "d", "u", CorrelationGuid), + new AlarmCommand("a", "Acknowledge", "u", "note", null), + ]; + + CommandEnvelope[] typed = + [ + MeshTransportDtoMapper.ToEnvelope((DispatchDeployment)commands[0]), + MeshTransportDtoMapper.ToEnvelope((RestartDriver)commands[1]), + MeshTransportDtoMapper.ToEnvelope((ReconnectDriver)commands[2]), + MeshTransportDtoMapper.ToEnvelope((AlarmCommand)commands[3], alarmCorrelation), + ]; + + for (var i = 0; i < commands.Length; i++) + { + MeshTransportDtoMapper.ToEnvelope(commands[i], alarmCorrelation).ShouldBe(typed[i]); + } + } +} From f8325373ef96d0206b6230b66fcae515da0f39c3 Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Thu, 6 Aug 2026 03:29:02 -0400 Subject: [PATCH 05/12] =?UTF-8?q?docs(plans):=20Task=200.2=20close-out=20?= =?UTF-8?q?=E2=80=94=20tracking=20+=20Phase=201=20dispatch=20notes?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- ...2026-08-06-clusterclient-to-grpc-tracking.md | 17 ++++++++++++++++- ...26-08-06-clusterclient-to-grpc.md.tasks.json | 2 +- 2 files changed, 17 insertions(+), 2 deletions(-) diff --git a/docs/plans/2026-08-06-clusterclient-to-grpc-tracking.md b/docs/plans/2026-08-06-clusterclient-to-grpc-tracking.md index 5bc95f15..20bb84fe 100644 --- a/docs/plans/2026-08-06-clusterclient-to-grpc-tracking.md +++ b/docs/plans/2026-08-06-clusterclient-to-grpc-tracking.md @@ -22,7 +22,7 @@ posture). | Phase | Scope | Branch | Status | PR / merge SHA | Live gate | |---|---|---|---|---|---| -| 0 | Proto + mappers + goldens; both auth interceptors; `IMeshPskProvider`; options/validators (+`Grpc` value, SplitTopology relaxation); rig dark config; stale-comment fixes; FleetStatus issue filed | `feat/grpc-phase0-contract-auth` | **IN PROGRESS** — Batch A (0.1 + 0.3) done @ `aa21f61f`, spec ✅ + code review ✅ (0 critical/important); 0.2 next | — | PENDING | +| 0 | Proto + mappers + goldens; both auth interceptors; `IMeshPskProvider`; options/validators (+`Grpc` value, SplitTopology relaxation); rig dark config; stale-comment fixes; FleetStatus issue filed | `feat/grpc-phase0-contract-auth` | **IN PROGRESS** — Batch A (0.1 + 0.3) done @ `aa21f61f` + close-out `09db9e47`; 0.2 (mapper + 27 wire-level goldens) done @ `545413a8`; all reviews spec ✅ + code ✅, 0 critical/important; Batch B (0.4 + 0.5) next | — | PENDING | | 1A | Central→node command plane (`NodeCommandService` on the Telemetry listener; `MeshGrpcCommandDialer`; third mode branch in `CentralCommunicationActor`) — **worktree A** | `feat/grpc-node-command` | NOT STARTED | — | PENDING | | 1B | Node→central ack plane (`CentralCommandService` on the ConfigServe listener; `GrpcApplyAckSender`; third branch in `NodeCommunicationActor`) — **worktree B**, rebases on 1A | `feat/grpc-central-ack` | NOT STARTED | — | PENDING | | 2 | Baseline reproduction on pre-flip build, then full-rig cutover (`OTOPCUA_MESH_MODE` default → `Grpc`) + soak + kill drills | (rig commit) | NOT STARTED | — | PENDING | @@ -48,6 +48,21 @@ phase DoD — PR + merge are user-owned.** | 2026-08-06 | `AlarmCommandMessage.operation` is a proto `string`, NOT an enum — the CLR `AlarmCommand.Operation` is an unconstrained string, and an enum would silently decode unknown operations to `UNSPECIFIED` (the drop class this migration removes). Plan T0.1's "AlarmCommand verb" enum wording was a plan defect; only `ApplyOutcome` carries the enum-0 rule. T0.2's goldens must pin pass-through of an unrecognized operation string | proto comment + this log (T0.1 review) | | 2026-08-06 | `correlation_id` lives on the envelope AND as inner fields on the three commands whose CLR record carries one — each command round-trips independently in the T0.2 mapper; the envelope covers `AlarmCommand` (no CLR correlation member). For the three, the inner field is authoritative on decode | proto comment + this log (T0.1 review) | +## Phase 1 dispatch notes (from Task 0.2 reviews — fold into the 1A/1B implementer prompts) + +- The alarm arm of `FromEnvelope` necessarily discards the envelope correlation + (`AlarmCommand` has no CLR member for it) — the 1A router must read + `envelope.CorrelationId` itself for logging. +- `ApplyAck` is the ONE record of the five that never rides `CommandEnvelope`: + `GrpcApplyAckSender` (1B) calls `MeshTransportDtoMapper.ToProto(ApplyAck)` directly on + `ReportApply` — do not expect (or add) a sixth `ToEnvelope` overload. +- Consider a thin try/catch at the `Execute`/`ReportApply` service boundary wrapping + `FromEnvelope`/`FromProto` failures with the envelope's correlation_id (readable as a plain + string even when inner ids fail to parse) so an operator can correlate a rejected call. +- Revisit the untyped `ToEnvelope(object, CorrelationId alarmCorrelationId)` signature when + the 1A router lands — if the router prefers one always-honoured transport correlation, + change the signature then, before it has more callers. + ## Defect log Defects found during execution (live-gate or review), with root cause + pinned test. diff --git a/docs/plans/2026-08-06-clusterclient-to-grpc.md.tasks.json b/docs/plans/2026-08-06-clusterclient-to-grpc.md.tasks.json index 2ffb2b35..ff6b445b 100644 --- a/docs/plans/2026-08-06-clusterclient-to-grpc.md.tasks.json +++ b/docs/plans/2026-08-06-clusterclient-to-grpc.md.tasks.json @@ -4,7 +4,7 @@ "worktrees": {}, "tasks": [ {"id": 0, "subject": "Task 0.1: mesh_transport.proto + contract lock", "phase": "0", "branch": "feat/grpc-phase0-contract-auth", "classification": "standard", "parallelizableWith": ["0.3 (batch, same agent)"], "status": "completed", "commit": "c0ccfa27", "review": "spec ✅ + code ✅"}, - {"id": 1, "subject": "Task 0.2: DTO mapper + round-trip goldens", "phase": "0", "branch": "feat/grpc-phase0-contract-auth", "classification": "high-risk", "parallelizableWith": [], "status": "in_progress", "blockedBy": [0]}, + {"id": 1, "subject": "Task 0.2: DTO mapper + round-trip goldens", "phase": "0", "branch": "feat/grpc-phase0-contract-auth", "classification": "high-risk", "parallelizableWith": [], "status": "completed", "blockedBy": [0], "commit": "545413a8", "review": "spec ✅ + code ✅ (serial)"}, {"id": 2, "subject": "Task 0.3: MeshTransportOptions + validators + SplitTopology relaxation", "phase": "0", "branch": "feat/grpc-phase0-contract-auth", "classification": "standard", "parallelizableWith": ["0.1 (batch, same agent)"], "status": "completed", "commit": "aa21f61f", "review": "spec ✅ + code ✅"}, {"id": 3, "subject": "Task 0.4: node-side MeshCommandAuthInterceptor", "phase": "0", "branch": "feat/grpc-phase0-contract-auth", "classification": "high-risk", "parallelizableWith": ["0.5 (batch, same agent)"], "status": "pending", "blockedBy": [2]}, {"id": 4, "subject": "Task 0.5: IMeshPskProvider + CentralCommandAuthInterceptor", "phase": "0", "branch": "feat/grpc-phase0-contract-auth", "classification": "high-risk", "parallelizableWith": ["0.4 (batch, same agent)"], "status": "pending", "blockedBy": [2]}, From fbda015c6e147e3afaf3c769fd0c836d067d2c7e Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Thu, 6 Aug 2026 03:37:02 -0400 Subject: [PATCH 06/12] feat(mesh): node-side PSK interceptor --- .../MeshCommandAuthInterceptor.cs | 173 +++++++++++++ src/Server/ZB.MOM.WW.OtOpcUa.Host/Program.cs | 7 + .../MeshCommandAuthInterceptorTests.cs | 227 ++++++++++++++++++ 3 files changed, 407 insertions(+) create mode 100644 src/Server/ZB.MOM.WW.OtOpcUa.Host/Configuration/MeshCommandAuthInterceptor.cs create mode 100644 tests/Server/ZB.MOM.WW.OtOpcUa.Host.Tests/Configuration/MeshCommandAuthInterceptorTests.cs diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.Host/Configuration/MeshCommandAuthInterceptor.cs b/src/Server/ZB.MOM.WW.OtOpcUa.Host/Configuration/MeshCommandAuthInterceptor.cs new file mode 100644 index 00000000..b7348d0e --- /dev/null +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Host/Configuration/MeshCommandAuthInterceptor.cs @@ -0,0 +1,173 @@ +using System.Security.Cryptography; +using System.Text; +using Grpc.Core; +using Grpc.Core.Interceptors; +using Microsoft.Extensions.Logging; +using Microsoft.Extensions.Options; +using ZB.MOM.WW.OtOpcUa.Cluster; + +namespace ZB.MOM.WW.OtOpcUa.Host.Configuration; + +/// +/// Gates the node-side command endpoint (ClusterClient→gRPC migration). The +/// NodeCommandService.Execute RPC is how central drives this node — restart a driver, +/// acknowledge an alarm, dispatch a deployment — so anything able to reach a node's command port +/// could otherwise operate the node. This interceptor is the ONLY inbound auth on that endpoint. +/// +/// +/// +/// Scoped by method path. Only calls under +/// /mesh_transport.v1.NodeCommandService/ are gated; every other method on the shared +/// gRPC pipeline (LocalDb sync, config-serve, telemetry) passes through untouched, so this can +/// share one AddGrpc registration with , +/// and . +/// +/// +/// Fail-closed. With no MeshTransport:GrpcPsk configured, NO command is +/// accepted, authenticated or not. Treating "no key" as "no auth required" would expose the +/// command plane on exactly the default shape every node ships with — and unlike the three +/// sibling endpoints, this one is a write surface, not a read one. Central must present the +/// same per-cluster key, so a key typo stops commands outright rather than degrading to +/// unauthenticated. +/// +/// +/// No x-otopcua-cluster requirement here, deliberately. That header is a +/// central-side routing hint — the caller's own claim about which cluster it is addressing. +/// Requiring a node to check it would gate on a value the caller chooses, which proves +/// nothing the key has not already proven; the key IS the cluster boundary, so a caller +/// holding another cluster's PSK is rejected by the comparison below. +/// +/// +/// Comparison is over UTF-8 bytes, so a +/// wrong key cannot be recovered byte-by-byte from response timing. Length differences are +/// unavoidably observable and are not sensitive. +/// +/// +/// Exactly one public constructor. Grpc.AspNetCore silently stops invoking an +/// interceptor that has more than one public ctor — see +/// MeshCommandAuthInterceptorTests.HasExactlyOnePublicConstructor. +/// +/// +public sealed class MeshCommandAuthInterceptor : Interceptor +{ + private const string ServicePrefix = "/mesh_transport.v1.NodeCommandService/"; + private const string AuthorizationHeader = "authorization"; + private const string BearerPrefix = "Bearer "; + + private readonly IOptions _options; + private readonly ILogger _logger; + + /// Creates the interceptor. + /// Mesh transport options; GrpcPsk is the expected bearer token. + /// Logger for denial diagnostics. + public MeshCommandAuthInterceptor( + IOptions options, + ILogger logger) + { + ArgumentNullException.ThrowIfNull(options); + ArgumentNullException.ThrowIfNull(logger); + + _options = options; + _logger = logger; + } + + /// + public override Task UnaryServerHandler( + TRequest request, + ServerCallContext context, + UnaryServerMethod continuation) + { + Authorize(context); + return continuation(request, context); + } + + /// + public override Task DuplexStreamingServerHandler( + IAsyncStreamReader requestStream, + IServerStreamWriter responseStream, + ServerCallContext context, + DuplexStreamingServerMethod continuation) + { + Authorize(context); + return continuation(requestStream, responseStream, context); + } + + /// + public override Task ClientStreamingServerHandler( + IAsyncStreamReader requestStream, + ServerCallContext context, + ClientStreamingServerMethod continuation) + { + Authorize(context); + return continuation(requestStream, context); + } + + /// + public override Task ServerStreamingServerHandler( + TRequest request, + IServerStreamWriter responseStream, + ServerCallContext context, + ServerStreamingServerMethod continuation) + { + Authorize(context); + return continuation(request, responseStream, context); + } + + /// + /// Throws with if this is + /// a node-command call that does not carry the configured bearer token. Non-command calls + /// return immediately. + /// + private void Authorize(ServerCallContext context) + { + if (!context.Method.StartsWith(ServicePrefix, StringComparison.Ordinal)) + return; + + var expected = _options.Value.GrpcPsk; + if (string.IsNullOrWhiteSpace(expected)) + { + _logger.LogWarning( + "Rejected a mesh command call to {Method}: no MeshTransport:GrpcPsk is configured, so " + + "the node command endpoint is closed. Set this node's per-cluster pre-shared key " + + "(MeshTransport__GrpcPsk), ideally as a ${{secret:OT-MESH-PSK-}} reference, " + + "and give central the same key for this cluster.", + context.Method); + throw new RpcException(new Status( + StatusCode.PermissionDenied, + "Node command service is not accepting calls: no pre-shared key is configured on this node.")); + } + + var presented = ExtractBearerToken(context.RequestHeaders); + if (presented is null || !FixedTimeEquals(presented, expected)) + { + _logger.LogWarning( + "Rejected a mesh command call to {Method}: {Reason}.", + context.Method, + presented is null ? "no bearer token presented" : "bearer token did not match"); + throw new RpcException(new Status( + StatusCode.PermissionDenied, + "Node command authentication failed.")); + } + } + + private static string? ExtractBearerToken(Metadata headers) + { + // gRPC lowercases header keys on the wire; compare case-insensitively anyway so a hand-built + // Metadata in a test behaves the same as a real request. + foreach (var entry in headers) + { + if (!string.Equals(entry.Key, AuthorizationHeader, StringComparison.OrdinalIgnoreCase)) + continue; + + var value = entry.Value; + if (value is not null && value.StartsWith(BearerPrefix, StringComparison.OrdinalIgnoreCase)) + return value[BearerPrefix.Length..]; + } + + return null; + } + + private static bool FixedTimeEquals(string presented, string expected) + => CryptographicOperations.FixedTimeEquals( + Encoding.UTF8.GetBytes(presented), Encoding.UTF8.GetBytes(expected)); +} diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.Host/Program.cs b/src/Server/ZB.MOM.WW.OtOpcUa.Host/Program.cs index cac7bbb6..1e84dfe8 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.Host/Program.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Host/Program.cs @@ -547,6 +547,12 @@ builder.Services.AddOtOpcUaObservability(builder.Configuration); // — the interceptor is the ONLY inbound auth on any endpoint (the libraries verify nothing). // Registered whenever either role is present so an admin-only node still serves config and a // driver-only node still gates sync + telemetry. +// +// MeshCommandAuthInterceptor (ClusterClient→gRPC migration) joins on the same terms: it gates the +// node-side NodeCommandService and is registered NOW, a pure pass-through until Phase 1 maps that +// service — an interceptor whose service is unmapped never sees a matching path. Registering it +// alongside the service it gates would put a security gate's wiring in the same commit as the write +// surface it protects; doing it here means the gate cannot be forgotten when that surface lights up. if (hasDriver || hasAdmin) { builder.Services.AddGrpc(o => @@ -555,6 +561,7 @@ if (hasDriver || hasAdmin) { o.Interceptors.Add(); o.Interceptors.Add(); + o.Interceptors.Add(); } if (hasAdmin) o.Interceptors.Add(); diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.Host.Tests/Configuration/MeshCommandAuthInterceptorTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.Host.Tests/Configuration/MeshCommandAuthInterceptorTests.cs new file mode 100644 index 00000000..d2446bbb --- /dev/null +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.Host.Tests/Configuration/MeshCommandAuthInterceptorTests.cs @@ -0,0 +1,227 @@ +using Grpc.Core; +using Microsoft.Extensions.Logging.Abstractions; +using Microsoft.Extensions.Options; +using Shouldly; +using Xunit; +using ZB.MOM.WW.OtOpcUa.Cluster; +using ZB.MOM.WW.OtOpcUa.Host.Configuration; + +namespace ZB.MOM.WW.OtOpcUa.Host.Tests.Configuration; + +/// +/// ClusterClient→gRPC Phase 0 — the node-side command endpoint's inbound gate. Mapping +/// NodeCommandService onto the pipeline is a Phase 1 task; this test covers only the +/// interceptor's own authorization behavior. +/// +/// +/// Same contract as , scoped to +/// /mesh_transport.v1.NodeCommandService/ — fail-closed on an unconfigured key, +/// constant-time comparison, gated on every handler shape so scoping holds regardless of RPC +/// shape. Deliberately NO x-otopcua-cluster requirement: that header is a central-side +/// routing hint, and requiring it here would gate on a value the caller chooses. +/// +public sealed class MeshCommandAuthInterceptorTests +{ + private const string ExecuteMethod = "/mesh_transport.v1.NodeCommandService/Execute"; + private const string OtherMethod = "/telemetry.v1.TelemetryStreamService/Subscribe"; + + private static MeshCommandAuthInterceptor CreateInterceptor(string? psk) + => new( + Options.Create(new MeshTransportOptions { GrpcPsk = psk }), + NullLogger.Instance); + + private static ServerCallContext CreateContext(string method, string? authorizationHeader) + { + var headers = new Metadata(); + if (authorizationHeader is not null) + headers.Add("authorization", authorizationHeader); + + return new FakeServerCallContext(method, headers); + } + + /// Invokes the interceptor's unary path (how Execute actually runs). + private static Task Invoke(MeshCommandAuthInterceptor interceptor, ServerCallContext context) + => interceptor.UnaryServerHandler( + "request", context, (_, _) => Task.FromResult("ok")); + + [Fact] + public async Task NonMeshCommandMethod_PassesThrough_EvenWithNoPskConfigured() + { + // The interceptor shares the AddGrpc pipeline with the sync / config-serve / telemetry + // services, so it sees every call. It must be scoped strictly to the node command service. + var interceptor = CreateInterceptor(psk: null); + var context = CreateContext(OtherMethod, authorizationHeader: null); + + await Invoke(interceptor, context); // no throw + } + + [Fact] + public async Task ExecuteMethod_WithNoPskConfigured_IsDenied_EvenWithABearerToken() + { + // Fail-closed. "No PSK configured" is the DEFAULT — treating it as "no auth required" would + // let anything that can reach the port drive RestartDriver / AlarmCommand / deployment + // dispatch on this node. A token must not help. + var interceptor = CreateInterceptor(psk: null); + var context = CreateContext(ExecuteMethod, "Bearer anything-at-all"); + + var ex = await Should.ThrowAsync(() => Invoke(interceptor, context)); + ex.StatusCode.ShouldBe(StatusCode.PermissionDenied); + } + + [Fact] + public async Task ExecuteMethod_WithBlankPskConfigured_IsDenied() + { + // Blank is distinct from unset in the options (GrpcPsk is nullable on purpose) but must be + // treated identically here: an empty expected token would be satisfied by an empty bearer. + var interceptor = CreateInterceptor(" "); + var context = CreateContext(ExecuteMethod, "Bearer "); + + var ex = await Should.ThrowAsync(() => Invoke(interceptor, context)); + ex.StatusCode.ShouldBe(StatusCode.PermissionDenied); + } + + [Fact] + public async Task ExecuteMethod_WithNoBearerToken_IsDenied() + { + var interceptor = CreateInterceptor("the-cluster-psk"); + var context = CreateContext(ExecuteMethod, authorizationHeader: null); + + var ex = await Should.ThrowAsync(() => Invoke(interceptor, context)); + ex.StatusCode.ShouldBe(StatusCode.PermissionDenied); + } + + [Fact] + public async Task ExecuteMethod_WithMalformedAuthorizationHeader_IsDenied() + { + // No "Bearer " prefix at all — must be treated the same as a missing header. + var interceptor = CreateInterceptor("the-cluster-psk"); + var context = CreateContext(ExecuteMethod, "the-cluster-psk"); + + var ex = await Should.ThrowAsync(() => Invoke(interceptor, context)); + ex.StatusCode.ShouldBe(StatusCode.PermissionDenied); + } + + [Fact] + public async Task ExecuteMethod_WithWrongBearerToken_IsDenied() + { + // This is also the isolation proof: a node holding SITE-A's PSK rejects a caller presenting + // SITE-B's. Nothing about the cluster id is consulted — only whether the key matches. + var interceptor = CreateInterceptor("dev-mesh-psk-SITE-A"); + var context = CreateContext(ExecuteMethod, "Bearer dev-mesh-psk-SITE-B"); + + var ex = await Should.ThrowAsync(() => Invoke(interceptor, context)); + ex.StatusCode.ShouldBe(StatusCode.PermissionDenied); + } + + [Fact] + public async Task ExecuteMethod_WithCorrectBearerToken_PassesThrough() + { + var interceptor = CreateInterceptor("the-cluster-psk"); + var context = CreateContext(ExecuteMethod, "Bearer the-cluster-psk"); + + await Invoke(interceptor, context); // no throw + } + + [Fact] + public async Task ExecuteMethod_WithCorrectBearerToken_AndNoClusterHeader_PassesThrough() + { + // x-otopcua-cluster is a central-side routing hint. Requiring it here would gate on a value + // the caller supplies about itself, which proves nothing the PSK has not already proven. + var interceptor = CreateInterceptor("the-cluster-psk"); + var context = CreateContext(ExecuteMethod, "Bearer the-cluster-psk"); + + await Invoke(interceptor, context); // no throw + } + + [Fact] + public async Task ExecuteMethod_TokenComparison_IsNotAPrefixMatch() + { + // A prefix/StartsWith comparison would accept a truncated key and make the secret + // recoverable one character at a time. + var interceptor = CreateInterceptor("the-cluster-psk"); + var context = CreateContext(ExecuteMethod, "Bearer the-cluster-ps"); + + var ex = await Should.ThrowAsync(() => Invoke(interceptor, context)); + ex.StatusCode.ShouldBe(StatusCode.PermissionDenied); + } + + [Fact] + public async Task ServerStreamingPath_IsAlsoGated() + { + // Execute is unary today, but gating only that path would leave any future streaming method + // on the same service open. Gate every handler shape. + var interceptor = CreateInterceptor("the-cluster-psk"); + var context = CreateContext(ExecuteMethod, "Bearer the-wrong-psk"); + + var ex = await Should.ThrowAsync(() => + interceptor.ServerStreamingServerHandler( + "request", responseStream: null!, context, (_, _, _) => Task.CompletedTask)); + + ex.StatusCode.ShouldBe(StatusCode.PermissionDenied); + } + + [Fact] + public async Task ClientStreamingPath_IsAlsoGated() + { + var interceptor = CreateInterceptor("the-cluster-psk"); + var context = CreateContext(ExecuteMethod, "Bearer the-wrong-psk"); + + var ex = await Should.ThrowAsync(() => + interceptor.ClientStreamingServerHandler( + requestStream: null!, context, (_, _) => Task.FromResult("ok"))); + + ex.StatusCode.ShouldBe(StatusCode.PermissionDenied); + } + + [Fact] + public async Task DuplexStreamingPath_IsAlsoGated() + { + var interceptor = CreateInterceptor("the-cluster-psk"); + var context = CreateContext(ExecuteMethod, "Bearer the-wrong-psk"); + + var ex = await Should.ThrowAsync(() => + interceptor.DuplexStreamingServerHandler( + requestStream: null!, responseStream: null!, context, (_, _, _) => Task.CompletedTask)); + + ex.StatusCode.ShouldBe(StatusCode.PermissionDenied); + } + + [Fact] + public void HasExactlyOnePublicConstructor() + { + // Grpc.AspNetCore silently stops invoking an interceptor with more than one public ctor — + // this pins the shape so a well-meaning overload doesn't quietly disable auth in production. + var publicCtors = typeof(MeshCommandAuthInterceptor).GetConstructors(); + + publicCtors.Length.ShouldBe(1); + } + + /// + /// Minimal carrying just a method name and request headers — + /// the only two things the interceptor reads. Hand-rolled because + /// Grpc.Core.Testing.TestServerCallContext ships only in the retired native package. + /// + private sealed class FakeServerCallContext(string method, Metadata requestHeaders) + : ServerCallContext + { + protected override string MethodCore => method; + protected override string HostCore => "localhost"; + protected override string PeerCore => "ipv4:127.0.0.1:12345"; + protected override DateTime DeadlineCore => DateTime.UtcNow.AddMinutes(1); + protected override Metadata RequestHeadersCore => requestHeaders; + protected override CancellationToken CancellationTokenCore => CancellationToken.None; + protected override Metadata ResponseTrailersCore { get; } = []; + protected override Status StatusCore { get; set; } + protected override WriteOptions? WriteOptionsCore { get; set; } + + protected override AuthContext AuthContextCore { get; } = + new(null, new Dictionary>()); + + protected override ContextPropagationToken CreatePropagationTokenCore( + ContextPropagationOptions? options) + => throw new NotSupportedException(); + + protected override Task WriteResponseHeadersAsyncCore(Metadata responseHeaders) + => Task.CompletedTask; + } +} From 68a416e4fc8e2c40e17a680e0491a49772476c6d Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Thu, 6 Aug 2026 03:43:05 -0400 Subject: [PATCH 07/12] feat(mesh): per-cluster PSK provider + central interceptor --- .../Communication/IMeshPskProvider.cs | 202 +++++++++++ .../ZB.MOM.WW.OtOpcUa.ControlPlane.csproj | 4 + .../CentralCommandAuthInterceptor.cs | 199 +++++++++++ src/Server/ZB.MOM.WW.OtOpcUa.Host/Program.cs | 24 +- .../Communication/MeshPskProviderTests.cs | 337 ++++++++++++++++++ .../CentralCommandAuthInterceptorTests.cs | 239 +++++++++++++ 6 files changed, 1000 insertions(+), 5 deletions(-) create mode 100644 src/Server/ZB.MOM.WW.OtOpcUa.ControlPlane/Communication/IMeshPskProvider.cs create mode 100644 src/Server/ZB.MOM.WW.OtOpcUa.Host/Configuration/CentralCommandAuthInterceptor.cs create mode 100644 tests/Server/ZB.MOM.WW.OtOpcUa.ControlPlane.Tests/Communication/MeshPskProviderTests.cs create mode 100644 tests/Server/ZB.MOM.WW.OtOpcUa.Host.Tests/Configuration/CentralCommandAuthInterceptorTests.cs diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.ControlPlane/Communication/IMeshPskProvider.cs b/src/Server/ZB.MOM.WW.OtOpcUa.ControlPlane/Communication/IMeshPskProvider.cs new file mode 100644 index 00000000..cbdbc987 --- /dev/null +++ b/src/Server/ZB.MOM.WW.OtOpcUa.ControlPlane/Communication/IMeshPskProvider.cs @@ -0,0 +1,202 @@ +using System.Collections.Concurrent; +using System.Diagnostics.CodeAnalysis; +using Microsoft.Extensions.Logging; +using Microsoft.Extensions.Options; +using ZB.MOM.WW.OtOpcUa.Cluster; +using ZB.MOM.WW.Secrets.Abstractions; + +namespace ZB.MOM.WW.OtOpcUa.ControlPlane.Communication; + +/// +/// Central's per-cluster pre-shared key lookup for the gRPC command plane. A node holds exactly +/// one key (MeshTransport:GrpcPsk, its own cluster's); central talks to every cluster, so +/// it needs one key per Cluster — both to authenticate itself when dialling a node and to +/// verify an inbound deployment ack. +/// +/// +/// +/// Fail-closed by contract. Every failure — unknown cluster, absent secret, blank +/// value, a store that throws — is reported as "no key" (false), and the caller +/// denies the inbound call or skips the outbound dial. There is no arm on which a caller may +/// proceed unauthenticated; a missing key must cost delivery, not authentication. +/// +/// +/// The Invalidate hooks exist because central's cluster set is DB-dynamic. They are +/// deliberately unconsumed at Phase 0 — Phase 1A wires them to the contact refresh, which is +/// the point at which central learns a cluster's membership (and so possibly its key) has +/// changed. +/// +/// +public interface IMeshPskProvider +{ + /// Gets the pre-shared key for one application Cluster. + /// The application Cluster id, matched case-insensitively. + /// The key, or null when none could be established. + /// true when a usable key was found; otherwise false. + bool TryGetPsk(string clusterId, [NotNullWhen(true)] out string? psk); + + /// Drops any cached key for one cluster, so the next lookup re-reads the store. + /// The application Cluster id, matched case-insensitively. + void Invalidate(string clusterId); + + /// Drops every cached key. + void InvalidateAll(); +} + +/// +public sealed class MeshPskProvider : IMeshPskProvider +{ + /// + /// Prefix of the secret holding a cluster's key — the full name is + /// OT-MESH-PSK-{clusterId}. The same name a node's + /// ${secret:OT-MESH-PSK-<clusterId>} reference expands, so central and the node + /// read one row rather than two conventions that drift. + /// + public const string SecretNamePrefix = "OT-MESH-PSK-"; + + private readonly IOptions _options; + private readonly ISecretResolver _secretResolver; + private readonly ILogger _logger; + + /// + /// Cached store resolutions, keyed by cluster id. Holds only values that came from the + /// Secrets store: the config map is re-read on every lookup (it is an in-memory dictionary), + /// which is what keeps the override unconditional rather than merely first-past-the-post. + /// + /// + /// under so a + /// burst of concurrent inbound acks for one cluster resolves the secret once instead of + /// stampeding the store. Negatives are never retained — see . + /// + private readonly ConcurrentDictionary> _cache = + new(StringComparer.OrdinalIgnoreCase); + + /// Creates the provider. + /// Mesh transport options; ClusterPsks is the override map. + /// The shared secret store read seam. + /// Logger for lookup-failure diagnostics. + public MeshPskProvider( + IOptions options, + ISecretResolver secretResolver, + ILogger logger) + { + ArgumentNullException.ThrowIfNull(options); + ArgumentNullException.ThrowIfNull(secretResolver); + ArgumentNullException.ThrowIfNull(logger); + + _options = options; + _secretResolver = secretResolver; + _logger = logger; + } + + /// + public bool TryGetPsk(string clusterId, [NotNullWhen(true)] out string? psk) + { + psk = null; + + if (string.IsNullOrWhiteSpace(clusterId)) + { + _logger.LogWarning( + "Mesh PSK lookup was asked for a blank cluster id; refusing rather than resolving a " + + "key that would belong to no cluster."); + return false; + } + + // Config map first, on every call. Blank counts as absent: an empty + // MeshTransport__ClusterPsks__ is an unset environment variable, not an operator + // deciding this cluster has an empty key. + if (_options.Value.ClusterPsks.TryGetValue(clusterId, out var configured) + && !string.IsNullOrWhiteSpace(configured)) + { + psk = configured; + return true; + } + + var entry = _cache.GetOrAdd( + clusterId, + static (id, self) => new Lazy( + () => self.ResolveFromStore(id), LazyThreadSafetyMode.ExecutionAndPublication), + this); + + var resolved = entry.Value; + if (string.IsNullOrWhiteSpace(resolved)) + { + // Never retain a negative. A store that was briefly unreachable would otherwise stay + // "no key" for this cluster until something called Invalidate — and at Phase 0 nothing + // does, so a transient fault would present as a permanent, silent command outage. + _cache.TryRemove(new KeyValuePair>(clusterId, entry)); + return false; + } + + psk = resolved; + return true; + } + + /// + public void Invalidate(string clusterId) + { + if (string.IsNullOrWhiteSpace(clusterId)) + return; + + _cache.TryRemove(clusterId, out _); + } + + /// + public void InvalidateAll() => _cache.Clear(); + + /// + /// Reads OT-MESH-PSK-{clusterId} from the Secrets store, converting every failure mode + /// into null. + /// + /// The application Cluster id. + /// The stored key, or null when it could not be established. + /// + /// + /// Blocking on the async resolver is deliberate. The consumer is a gRPC server + /// interceptor, whose Authorize is synchronous in all four sibling interceptors, + /// and making this seam async would push that shape onto every one of them. The block + /// happens at most once per cluster per process — every later lookup is a dictionary hit + /// — and the store is a node-local SQLite file, not a network call. + /// + /// + /// The catch is broad on purpose: this is an authentication boundary, and the set of + /// exceptions a secret store can raise (missing KEK, corrupt row, IO) is neither closed + /// nor interesting here. Anything other than a usable key is "no key". + /// + /// + private string? ResolveFromStore(string clusterId) + { + SecretName name; + try + { + name = new SecretName(SecretNamePrefix + clusterId); + } + catch (ArgumentException ex) + { + // Cluster ids are operator-authored and may carry characters SecretName's allow-list + // rejects. Log the id — it is not sensitive — and never the key. + _logger.LogWarning( + ex, + "Cluster {ClusterId} cannot be expressed as a secret name, so its mesh PSK cannot be " + + "resolved from the store. Supply it through MeshTransport:ClusterPsks instead, or " + + "rename the cluster to [a-z0-9._/-].", + clusterId); + return null; + } + + try + { + return _secretResolver.GetAsync(name, CancellationToken.None).GetAwaiter().GetResult(); + } + catch (Exception ex) + { + _logger.LogWarning( + ex, + "Failed to resolve the mesh PSK for cluster {ClusterId} from secret {SecretName}. " + + "Commands to that cluster and acks from it are refused until it resolves.", + clusterId, + name.Value); + return null; + } + } +} diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.ControlPlane/ZB.MOM.WW.OtOpcUa.ControlPlane.csproj b/src/Server/ZB.MOM.WW.OtOpcUa.ControlPlane/ZB.MOM.WW.OtOpcUa.ControlPlane.csproj index 89d2622c..0ffdd314 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.ControlPlane/ZB.MOM.WW.OtOpcUa.ControlPlane.csproj +++ b/src/Server/ZB.MOM.WW.OtOpcUa.ControlPlane/ZB.MOM.WW.OtOpcUa.ControlPlane.csproj @@ -19,6 +19,10 @@ Google.Protobuf + Grpc.Core.Api flow transitively from Commons (the generated client). --> + +