feat(mesh): per-cluster PSK provider + central interceptor

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;
+
+/// <summary>
+///     Central's per-cluster pre-shared key lookup for the gRPC command plane. A node holds exactly
+///     one key (<c>MeshTransport:GrpcPsk</c>, its own cluster's); central talks to every cluster, so
+///     it needs one key per <c>Cluster</c> — both to authenticate itself when dialling a node and to
+///     verify an inbound deployment ack.
+/// </summary>
+/// <remarks>
+///     <para>
+///         <b>Fail-closed by contract.</b> Every failure — unknown cluster, absent secret, blank
+///         value, a store that throws — is reported as "no key" (<c>false</c>), 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.
+///     </para>
+///     <para>
+///         The <c>Invalidate</c> 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.
+///     </para>
+/// </remarks>
+public interface IMeshPskProvider
+{
+    /// <summary>Gets the pre-shared key for one application <c>Cluster</c>.</summary>
+    /// <param name="clusterId">The application <c>Cluster</c> id, matched case-insensitively.</param>
+    /// <param name="psk">The key, or <c>null</c> when none could be established.</param>
+    /// <returns><c>true</c> when a usable key was found; otherwise <c>false</c>.</returns>
+    bool TryGetPsk(string clusterId, [NotNullWhen(true)] out string? psk);
+
+    /// <summary>Drops any cached key for one cluster, so the next lookup re-reads the store.</summary>
+    /// <param name="clusterId">The application <c>Cluster</c> id, matched case-insensitively.</param>
+    void Invalidate(string clusterId);
+
+    /// <summary>Drops every cached key.</summary>
+    void InvalidateAll();
+}
+
+/// <inheritdoc />
+public sealed class MeshPskProvider : IMeshPskProvider
+{
+    /// <summary>
+    ///     Prefix of the secret holding a cluster's key — the full name is
+    ///     <c>OT-MESH-PSK-{clusterId}</c>. The same name a node's
+    ///     <c>${secret:OT-MESH-PSK-&lt;clusterId&gt;}</c> reference expands, so central and the node
+    ///     read one row rather than two conventions that drift.
+    /// </summary>
+    public const string SecretNamePrefix = "OT-MESH-PSK-";
+
+    private readonly IOptions<MeshTransportOptions> _options;
+    private readonly ISecretResolver _secretResolver;
+    private readonly ILogger<MeshPskProvider> _logger;
+
+    /// <summary>
+    ///     Cached store resolutions, keyed by cluster id. Holds <b>only</b> 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.
+    /// </summary>
+    /// <remarks>
+    ///     <see cref="Lazy{T}"/> under <see cref="LazyThreadSafetyMode.ExecutionAndPublication"/> so a
+    ///     burst of concurrent inbound acks for one cluster resolves the secret once instead of
+    ///     stampeding the store. Negatives are never retained — see <see cref="TryGetPsk"/>.
+    /// </remarks>
+    private readonly ConcurrentDictionary<string, Lazy<string?>> _cache =
+        new(StringComparer.OrdinalIgnoreCase);
+
+    /// <summary>Creates the provider.</summary>
+    /// <param name="options">Mesh transport options; <c>ClusterPsks</c> is the override map.</param>
+    /// <param name="secretResolver">The shared secret store read seam.</param>
+    /// <param name="logger">Logger for lookup-failure diagnostics.</param>
+    public MeshPskProvider(
+        IOptions<MeshTransportOptions> options,
+        ISecretResolver secretResolver,
+        ILogger<MeshPskProvider> logger)
+    {
+        ArgumentNullException.ThrowIfNull(options);
+        ArgumentNullException.ThrowIfNull(secretResolver);
+        ArgumentNullException.ThrowIfNull(logger);
+
+        _options = options;
+        _secretResolver = secretResolver;
+        _logger = logger;
+    }
+
+    /// <inheritdoc />
+    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__<id> 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<string?>(
+                () => 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<string, Lazy<string?>>(clusterId, entry));
+            return false;
+        }
+
+        psk = resolved;
+        return true;
+    }
+
+    /// <inheritdoc />
+    public void Invalidate(string clusterId)
+    {
+        if (string.IsNullOrWhiteSpace(clusterId))
+            return;
+
+        _cache.TryRemove(clusterId, out _);
+    }
+
+    /// <inheritdoc />
+    public void InvalidateAll() => _cache.Clear();
+
+    /// <summary>
+    ///     Reads <c>OT-MESH-PSK-{clusterId}</c> from the Secrets store, converting every failure mode
+    ///     into <c>null</c>.
+    /// </summary>
+    /// <param name="clusterId">The application <c>Cluster</c> id.</param>
+    /// <returns>The stored key, or <c>null</c> when it could not be established.</returns>
+    /// <remarks>
+    ///     <para>
+    ///         Blocking on the async resolver is deliberate. The consumer is a gRPC server
+    ///         interceptor, whose <c>Authorize</c> 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.
+    ///     </para>
+    ///     <para>
+    ///         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".
+    ///     </para>
+    /// </remarks>
+    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;
+        }
+    }
+}

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). -->
         <PackageReference Include="Grpc.Net.Client"/>
         <PackageReference Include="ZB.MOM.WW.Audit"/>
+        <!-- ClusterClient→gRPC: MeshPskProvider resolves OT-MESH-PSK-{clusterId} from the shared
+             secret store when MeshTransport:ClusterPsks carries no override. Abstractions only —
+             the concrete store is the host's choice (SecretsRegistration), not the control plane's. -->
+        <PackageReference Include="ZB.MOM.WW.Secrets.Abstractions"/>
         <!-- For ClusterActiveNode — the shared "oldest Up member of role X" primitive. The control
              plane and the /health/active tier must give the SAME answer for which node is in charge,
              or the tier reports one node active while the Primary-gated data plane runs on another;

src/Server/ZB.MOM.WW.OtOpcUa.Host/Configuration/CentralCommandAuthInterceptor.cs

@@ -0,0 +1,199 @@
+using System.Security.Cryptography;
+using System.Text;
+using Grpc.Core;
+using Grpc.Core.Interceptors;
+using Microsoft.Extensions.Logging;
+using ZB.MOM.WW.OtOpcUa.ControlPlane.Communication;
+
+namespace ZB.MOM.WW.OtOpcUa.Host.Configuration;
+
+/// <summary>
+///     Gates the central-side command endpoint (ClusterClient→gRPC migration). The
+///     <c>CentralCommandService.ReportApply</c> RPC is how a node tells central its deployment
+///     applied; the coordinator counts those acks, so anything able to reach central's command port
+///     could otherwise forge a fleet-wide "everything applied". This interceptor is the ONLY inbound
+///     auth on that endpoint.
+/// </summary>
+/// <remarks>
+///     <para>
+///         <b>Scoped by method path.</b> Only calls under
+///         <c>/mesh_transport.v1.CentralCommandService/</c> are gated; every other method on the
+///         shared gRPC pipeline (config-serve, and on a fused node sync + telemetry + the node
+///         command service) passes through untouched, so this shares one <c>AddGrpc</c> registration
+///         with its siblings.
+///     </para>
+///     <para>
+///         <b>Per-cluster key, not one server key.</b> Unlike the four sibling interceptors there is
+///         no single expected token: central talks to N site clusters, each with its own PSK. The
+///         caller names its cluster in <c>x-otopcua-cluster</c> and the bearer is verified against
+///         <i>that</i> cluster's key via <see cref="IMeshPskProvider"/>. The header is therefore a
+///         claim the caller must then prove, not a credential — choosing a different cluster only
+///         changes which key it is graded against. A single fleet-wide key would let one compromised
+///         site node forge acks on behalf of every other site.
+///     </para>
+///     <para>
+///         <b>Fail-closed.</b> A missing or blank cluster header, and a cluster whose key cannot be
+///         established (unseeded, blank, or a secret store that is down), are both denied. There is
+///         deliberately no "try every known key" fallback: it would restore exactly the fleet-wide
+///         key this design exists to avoid.
+///     </para>
+///     <para>
+///         Comparison is <see cref="CryptographicOperations.FixedTimeEquals"/> 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.
+///     </para>
+///     <para>
+///         <b>All three denials return the same status text</b>, unlike the sibling interceptors,
+///         which distinguish "no key configured" from "authentication failed". Here that distinction
+///         would be an enumeration oracle: the caller supplies the cluster id, so a distinguishable
+///         "no key for that cluster" reply would let an unauthenticated caller map which clusters
+///         central holds keys for. The operator-facing detail goes to the log instead.
+///     </para>
+///     <para>
+///         <b>Exactly one public constructor.</b> <c>Grpc.AspNetCore</c> silently stops invoking an
+///         interceptor that has more than one public ctor — see
+///         <c>CentralCommandAuthInterceptorTests.HasExactlyOnePublicConstructor</c>.
+///     </para>
+/// </remarks>
+public sealed class CentralCommandAuthInterceptor : Interceptor
+{
+    private const string ServicePrefix = "/mesh_transport.v1.CentralCommandService/";
+    private const string AuthorizationHeader = "authorization";
+    private const string ClusterHeader = "x-otopcua-cluster";
+    private const string BearerPrefix = "Bearer ";
+
+    private readonly IMeshPskProvider _pskProvider;
+    private readonly ILogger<CentralCommandAuthInterceptor> _logger;
+
+    /// <summary>Creates the interceptor.</summary>
+    /// <param name="pskProvider">Supplies the expected bearer token for a given cluster.</param>
+    /// <param name="logger">Logger for denial diagnostics.</param>
+    public CentralCommandAuthInterceptor(
+        IMeshPskProvider pskProvider,
+        ILogger<CentralCommandAuthInterceptor> logger)
+    {
+        ArgumentNullException.ThrowIfNull(pskProvider);
+        ArgumentNullException.ThrowIfNull(logger);
+
+        _pskProvider = pskProvider;
+        _logger = logger;
+    }
+
+    /// <inheritdoc />
+    public override Task<TResponse> UnaryServerHandler<TRequest, TResponse>(
+        TRequest request,
+        ServerCallContext context,
+        UnaryServerMethod<TRequest, TResponse> continuation)
+    {
+        Authorize(context);
+        return continuation(request, context);
+    }
+
+    /// <inheritdoc />
+    public override Task DuplexStreamingServerHandler<TRequest, TResponse>(
+        IAsyncStreamReader<TRequest> requestStream,
+        IServerStreamWriter<TResponse> responseStream,
+        ServerCallContext context,
+        DuplexStreamingServerMethod<TRequest, TResponse> continuation)
+    {
+        Authorize(context);
+        return continuation(requestStream, responseStream, context);
+    }
+
+    /// <inheritdoc />
+    public override Task<TResponse> ClientStreamingServerHandler<TRequest, TResponse>(
+        IAsyncStreamReader<TRequest> requestStream,
+        ServerCallContext context,
+        ClientStreamingServerMethod<TRequest, TResponse> continuation)
+    {
+        Authorize(context);
+        return continuation(requestStream, context);
+    }
+
+    /// <inheritdoc />
+    public override Task ServerStreamingServerHandler<TRequest, TResponse>(
+        TRequest request,
+        IServerStreamWriter<TResponse> responseStream,
+        ServerCallContext context,
+        ServerStreamingServerMethod<TRequest, TResponse> continuation)
+    {
+        Authorize(context);
+        return continuation(request, responseStream, context);
+    }
+
+    /// <summary>
+    ///     Throws <see cref="RpcException"/> with <see cref="StatusCode.PermissionDenied"/> if this is
+    ///     a central-command call that does not name a cluster, names one with no resolvable key, or
+    ///     does not carry that cluster's bearer token. Non-command calls return immediately.
+    /// </summary>
+    private void Authorize(ServerCallContext context)
+    {
+        if (!context.Method.StartsWith(ServicePrefix, StringComparison.Ordinal))
+            return;
+
+        var clusterId = ExtractHeader(context.RequestHeaders, ClusterHeader);
+        if (string.IsNullOrWhiteSpace(clusterId))
+        {
+            _logger.LogWarning(
+                "Rejected a central command call to {Method}: no {Header} header, so there is no "
+                + "cluster whose pre-shared key the caller could be graded against.",
+                context.Method,
+                ClusterHeader);
+            throw new RpcException(new Status(
+                StatusCode.PermissionDenied,
+                "Central command authentication failed."));
+        }
+
+        if (!_pskProvider.TryGetPsk(clusterId, out var expected))
+        {
+            _logger.LogWarning(
+                "Rejected a central command call to {Method} claiming cluster {ClusterId}: no "
+                + "pre-shared key could be established for it. Seed the secret "
+                + "OT-MESH-PSK-<clusterId> on this node, or set "
+                + "MeshTransport:ClusterPsks:<clusterId> on a rig.",
+                context.Method,
+                clusterId);
+            throw new RpcException(new Status(
+                StatusCode.PermissionDenied,
+                "Central command authentication failed."));
+        }
+
+        var presented = ExtractBearerToken(context.RequestHeaders);
+        if (presented is null || !FixedTimeEquals(presented, expected))
+        {
+            _logger.LogWarning(
+                "Rejected a central command call to {Method} claiming cluster {ClusterId}: {Reason}.",
+                context.Method,
+                clusterId,
+                presented is null ? "no bearer token presented" : "bearer token did not match");
+            throw new RpcException(new Status(
+                StatusCode.PermissionDenied,
+                "Central command authentication failed."));
+        }
+    }
+
+    private static string? ExtractBearerToken(Metadata headers)
+    {
+        var value = ExtractHeader(headers, AuthorizationHeader);
+        return value is not null && value.StartsWith(BearerPrefix, StringComparison.OrdinalIgnoreCase)
+            ? value[BearerPrefix.Length..]
+            : null;
+    }
+
+    private static string? ExtractHeader(Metadata headers, string name)
+    {
+        // 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, name, StringComparison.OrdinalIgnoreCase))
+                return entry.Value;
+        }
+
+        return null;
+    }
+
+    private static bool FixedTimeEquals(string presented, string expected)
+        => CryptographicOperations.FixedTimeEquals(
+            Encoding.UTF8.GetBytes(presented), Encoding.UTF8.GetBytes(expected));
+}

src/Server/ZB.MOM.WW.OtOpcUa.Host/Program.cs

@@ -18,6 +18,7 @@ using ZB.MOM.WW.OtOpcUa.Commons.Interfaces;
 using ZB.MOM.WW.OtOpcUa.Configuration;
 using ZB.MOM.WW.OtOpcUa.Configuration.Services;
 using ZB.MOM.WW.OtOpcUa.ControlPlane;
+using ZB.MOM.WW.OtOpcUa.ControlPlane.Communication;
 using ZB.MOM.WW.OtOpcUa.ControlPlane.Redundancy;
 using ZB.MOM.WW.OtOpcUa.Commons.OpcUa;
 using ZB.MOM.WW.OtOpcUa.Commons.Engines;
@@ -526,6 +527,13 @@ if (hasAdmin)
     // hosted service has built the system).
     builder.Services.TryAddSingleton<Func<ActorSystem>>(sp => () => sp.GetRequiredService<ActorSystem>());
     builder.Services.AddSingleton<IManualFailoverService, ManualFailoverService>();
+    // Per-cluster mesh PSKs (ClusterClient→gRPC). Admin-only, and singleton because its whole job is
+    // to hold a cache: central talks to N clusters and would otherwise re-read the secret store on
+    // every inbound ack. A node needs nothing of this — it holds exactly its own MeshTransport:GrpcPsk
+    // — so registering it fleet-wide would give a driver-only node a lookup it can never use.
+    // Depends on ISecretResolver, which AddOtOpcUaSecrets registers unconditionally below (DI
+    // registration order is irrelevant; resolution order is not).
+    builder.Services.AddSingleton<IMeshPskProvider, MeshPskProvider>();
 }
 
 // Registered unconditionally: driver-role nodes resolve Layer-B DriverConfig secrets and have no auth/DP/AdminUI.
@@ -548,11 +556,14 @@ builder.Services.AddOtOpcUaObservability(builder.Configuration);
 // 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.
+// The two mesh-command interceptors (ClusterClient→gRPC migration) join on the same terms:
+// MeshCommandAuthInterceptor gates the node-side NodeCommandService (driver-role, one key — this
+// node's own), CentralCommandAuthInterceptor gates the central-side CentralCommandService
+// (admin-role, one key per cluster via IMeshPskProvider). Both are registered NOW and are pure
+// pass-throughs until Phase 1 maps those services — an interceptor whose service is unmapped never
+// sees a matching path. Registering them alongside the services they gate 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 =>
@@ -564,7 +575,10 @@ if (hasDriver || hasAdmin)
             o.Interceptors.Add<MeshCommandAuthInterceptor>();
         }
         if (hasAdmin)
+        {
             o.Interceptors.Add<ConfigServeAuthInterceptor>();
+            o.Interceptors.Add<CentralCommandAuthInterceptor>();
+        }
     });
 }