dohertj2/lmxopcua
1
0
Fork 0
Code Issues 14 Pull Requests Actions Packages Projects Releases Wiki Activity

feat(mesh): node-side PSK interceptor

Browse Source
This commit is contained in:
Joseph Doherty
2026-08-06 03:37:02 -04:00
parent f8325373ef
commit fbda015c6e
3 changed files with 407 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/Server/ZB.MOM.WW.OtOpcUa.Host/Configuration/MeshCommandAuthInterceptor.cs
+173
View File
@@ -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;
/// <summary>
/// Gates the node-side command endpoint (ClusterClient→gRPC migration). The
/// <c>NodeCommandService.Execute</c> 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.
/// </summary>
/// <remarks>
/// <para>
/// <b>Scoped by method path.</b> Only calls under
/// <c>/mesh_transport.v1.NodeCommandService/</c> are gated; every other method on the shared
/// gRPC pipeline (LocalDb sync, config-serve, telemetry) passes through untouched, so this can
/// share one <c>AddGrpc</c> registration with <see cref="LocalDbSyncAuthInterceptor"/>,
/// <see cref="ConfigServeAuthInterceptor"/> and <see cref="TelemetryStreamAuthInterceptor"/>.
/// </para>
/// <para>
/// <b>Fail-closed.</b> With no <c>MeshTransport:GrpcPsk</c> 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.
/// </para>
/// <para>
/// <b>No <c>x-otopcua-cluster</c> requirement here, deliberately.</b> 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.
/// </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>Exactly one public constructor.</b> <c>Grpc.AspNetCore</c> silently stops invoking an
/// interceptor that has more than one public ctor — see
/// <c>MeshCommandAuthInterceptorTests.HasExactlyOnePublicConstructor</c>.
/// </para>
/// </remarks>
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<MeshTransportOptions> _options;
private readonly ILogger<MeshCommandAuthInterceptor> _logger;
/// <summary>Creates the interceptor.</summary>
/// <param name="options">Mesh transport options; <c>GrpcPsk</c> is the expected bearer token.</param>
/// <param name="logger">Logger for denial diagnostics.</param>
public MeshCommandAuthInterceptor(
IOptions<MeshTransportOptions> options,
ILogger<MeshCommandAuthInterceptor> logger)
{
ArgumentNullException.ThrowIfNull(options);
ArgumentNullException.ThrowIfNull(logger);
_options = options;
_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 node-command call that does not carry the configured bearer token. Non-command calls
/// return immediately.
/// </summary>
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-<clusterId>}} 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));
}
src/Server/ZB.MOM.WW.OtOpcUa.Host/Program.cs
+7
View File
@@ -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<LocalDbSyncAuthInterceptor>();
o.Interceptors.Add<TelemetryStreamAuthInterceptor>();
o.Interceptors.Add<MeshCommandAuthInterceptor>();
}
if (hasAdmin)
o.Interceptors.Add<ConfigServeAuthInterceptor>();
tests/Server/ZB.MOM.WW.OtOpcUa.Host.Tests/Configuration/MeshCommandAuthInterceptorTests.cs
+227
View File
@@ -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;
/// <summary>
/// ClusterClient→gRPC Phase 0 — the node-side command endpoint's inbound gate. Mapping
/// <c>NodeCommandService</c> onto the pipeline is a Phase 1 task; this test covers only the
/// interceptor's own authorization behavior.
/// </summary>
/// <remarks>
/// Same contract as <see cref="TelemetryStreamAuthInterceptor"/>, scoped to
/// <c>/mesh_transport.v1.NodeCommandService/</c> — fail-closed on an unconfigured key,
/// constant-time comparison, gated on every handler shape so scoping holds regardless of RPC
/// shape. Deliberately NO <c>x-otopcua-cluster</c> requirement: that header is a central-side
/// routing hint, and requiring it here would gate on a value the caller chooses.
/// </remarks>
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<MeshCommandAuthInterceptor>.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);
}
/// <summary>Invokes the interceptor's unary path (how <c>Execute</c> actually runs).</summary>
private static Task Invoke(MeshCommandAuthInterceptor interceptor, ServerCallContext context)
=> interceptor.UnaryServerHandler<string, string>(
"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<RpcException>(() => 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<RpcException>(() => 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<RpcException>(() => 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<RpcException>(() => 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<RpcException>(() => 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<RpcException>(() => 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<RpcException>(() =>
interceptor.ServerStreamingServerHandler<string, string>(
"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<RpcException>(() =>
interceptor.ClientStreamingServerHandler<string, string>(
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<RpcException>(() =>
interceptor.DuplexStreamingServerHandler<string, string>(
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);
}
/// <summary>
/// Minimal <see cref="ServerCallContext"/> carrying just a method name and request headers —
/// the only two things the interceptor reads. Hand-rolled because
/// <c>Grpc.Core.Testing.TestServerCallContext</c> ships only in the retired native package.
/// </summary>
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<string, List<AuthProperty>>());
protected override ContextPropagationToken CreatePropagationTokenCore(
ContextPropagationOptions? options)
=> throw new NotSupportedException();
protected override Task WriteResponseHeadersAsyncCore(Metadata responseHeaders)
=> Task.CompletedTask;
}
}