fix(mesh): Batch B review fixes — bounded store read, cache poison guard, test hardening
This commit is contained in:
@@ -26,6 +26,14 @@ namespace ZB.MOM.WW.OtOpcUa.ControlPlane.Communication;
|
|||||||
/// the point at which central learns a cluster's membership (and so possibly its key) has
|
/// the point at which central learns a cluster's membership (and so possibly its key) has
|
||||||
/// changed.
|
/// changed.
|
||||||
/// </para>
|
/// </para>
|
||||||
|
/// <para>
|
||||||
|
/// <b><c>Invalidate</c> is not a revocation barrier.</b> It drops the cached entry; it does
|
||||||
|
/// not wait for, cancel, or supersede a resolve already in flight. A lookup that entered the
|
||||||
|
/// store before the call can still publish and return the pre-rotation key to its caller, and
|
||||||
|
/// a lookup racing the eviction can re-cache the old value it had already read. Key rotation
|
||||||
|
/// is therefore eventually consistent by seconds, not atomic — treat <c>Invalidate</c> as
|
||||||
|
/// "stop serving this from memory", never as "this key is now dead everywhere".
|
||||||
|
/// </para>
|
||||||
/// </remarks>
|
/// </remarks>
|
||||||
public interface IMeshPskProvider
|
public interface IMeshPskProvider
|
||||||
{
|
{
|
||||||
@@ -54,6 +62,14 @@ public sealed class MeshPskProvider : IMeshPskProvider
|
|||||||
/// </summary>
|
/// </summary>
|
||||||
public const string SecretNamePrefix = "OT-MESH-PSK-";
|
public const string SecretNamePrefix = "OT-MESH-PSK-";
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// Ceiling on one store read. Not a tuning knob — a backstop: the read runs under the
|
||||||
|
/// <see cref="Lazy{T}"/> publication lock, so a store that never answers would park the
|
||||||
|
/// calling gRPC thread <i>and</i> every other caller queued behind it for that cluster.
|
||||||
|
/// Expiry fails closed like any other fault.
|
||||||
|
/// </summary>
|
||||||
|
private static readonly TimeSpan StoreReadTimeout = TimeSpan.FromSeconds(3);
|
||||||
|
|
||||||
private readonly IOptions<MeshTransportOptions> _options;
|
private readonly IOptions<MeshTransportOptions> _options;
|
||||||
private readonly ISecretResolver _secretResolver;
|
private readonly ISecretResolver _secretResolver;
|
||||||
private readonly ILogger<MeshPskProvider> _logger;
|
private readonly ILogger<MeshPskProvider> _logger;
|
||||||
@@ -118,7 +134,27 @@ public sealed class MeshPskProvider : IMeshPskProvider
|
|||||||
() => self.ResolveFromStore(id), LazyThreadSafetyMode.ExecutionAndPublication),
|
() => self.ResolveFromStore(id), LazyThreadSafetyMode.ExecutionAndPublication),
|
||||||
this);
|
this);
|
||||||
|
|
||||||
var resolved = entry.Value;
|
string? resolved;
|
||||||
|
try
|
||||||
|
{
|
||||||
|
resolved = entry.Value;
|
||||||
|
}
|
||||||
|
catch
|
||||||
|
{
|
||||||
|
// ResolveFromStore swallows everything the STORE can throw — but it logs while doing so,
|
||||||
|
// and a logger provider that throws escapes the factory. Lazy under
|
||||||
|
// ExecutionAndPublication caches that exception and rethrows it on every later read, so
|
||||||
|
// without this guard one bad log call would deny the cluster for the life of the process
|
||||||
|
// (nothing calls Invalidate until Phase 1A). Evict instance-keyed, same as the negative
|
||||||
|
// path below, and fail closed.
|
||||||
|
//
|
||||||
|
// Deliberately silent: the most likely reason we are here is that logging is what threw,
|
||||||
|
// so logging the failure is the one action with a real chance of throwing again — out of
|
||||||
|
// an authentication gate, turning a denial into a 500.
|
||||||
|
_cache.TryRemove(new KeyValuePair<string, Lazy<string?>>(clusterId, entry));
|
||||||
|
return false;
|
||||||
|
}
|
||||||
|
|
||||||
if (string.IsNullOrWhiteSpace(resolved))
|
if (string.IsNullOrWhiteSpace(resolved))
|
||||||
{
|
{
|
||||||
// Never retain a negative. A store that was briefly unreachable would otherwise stay
|
// Never retain a negative. A store that was briefly unreachable would otherwise stay
|
||||||
@@ -154,9 +190,12 @@ public sealed class MeshPskProvider : IMeshPskProvider
|
|||||||
/// <para>
|
/// <para>
|
||||||
/// Blocking on the async resolver is deliberate. The consumer is a gRPC server
|
/// 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,
|
/// 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
|
/// and making this seam async would push that shape onto every one of them. What makes
|
||||||
/// happens at most once per cluster per process — every later lookup is a dictionary hit
|
/// the block acceptable is that it is <b>bounded</b> (<see cref="StoreReadTimeout"/>) and
|
||||||
/// — and the store is a node-local SQLite file, not a network call.
|
/// happens at most once per cluster per process — every later lookup is a dictionary hit.
|
||||||
|
/// The bound does not assume anything about what backs the store: this code cannot see
|
||||||
|
/// whether the resolver reads a local file or crosses a network, so it refuses to wait
|
||||||
|
/// forever on either.
|
||||||
/// </para>
|
/// </para>
|
||||||
/// <para>
|
/// <para>
|
||||||
/// The catch is broad on purpose: this is an authentication boundary, and the set of
|
/// The catch is broad on purpose: this is an authentication boundary, and the set of
|
||||||
@@ -184,12 +223,15 @@ public sealed class MeshPskProvider : IMeshPskProvider
|
|||||||
return null;
|
return null;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
using var cts = new CancellationTokenSource(StoreReadTimeout);
|
||||||
try
|
try
|
||||||
{
|
{
|
||||||
return _secretResolver.GetAsync(name, CancellationToken.None).GetAwaiter().GetResult();
|
return _secretResolver.GetAsync(name, cts.Token).GetAwaiter().GetResult();
|
||||||
}
|
}
|
||||||
catch (Exception ex)
|
catch (Exception ex)
|
||||||
{
|
{
|
||||||
|
// Covers the timeout too — a cancelled read arrives here as OperationCanceledException
|
||||||
|
// and is treated as every other fault is: no key.
|
||||||
_logger.LogWarning(
|
_logger.LogWarning(
|
||||||
ex,
|
ex,
|
||||||
"Failed to resolve the mesh PSK for cluster {ClusterId} from secret {SecretName}. "
|
"Failed to resolve the mesh PSK for cluster {ClusterId} from secret {SecretName}. "
|
||||||
|
|||||||
@@ -45,9 +45,18 @@ namespace ZB.MOM.WW.OtOpcUa.Host.Configuration;
|
|||||||
/// <para>
|
/// <para>
|
||||||
/// <b>All three denials return the same status text</b>, unlike the sibling interceptors,
|
/// <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
|
/// 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
|
/// would be a <i>content</i> oracle: the caller supplies the cluster id, so a reply that said
|
||||||
/// "no key for that cluster" reply would let an unauthenticated caller map which clusters
|
/// "no key for that cluster" would let an unauthenticated caller read central's cluster set
|
||||||
/// central holds keys for. The operator-facing detail goes to the log instead.
|
/// straight off the wire. The operator-facing detail goes to the log instead.
|
||||||
|
/// </para>
|
||||||
|
/// <para>
|
||||||
|
/// <b>That is not yet full enumeration resistance, and should not be read as such.</b> A
|
||||||
|
/// timing oracle survives it: an unknown cluster misses the provider's cache and pays a store
|
||||||
|
/// read on every probe, while a known one is answered from memory after the first. Closing
|
||||||
|
/// that needs the lookup to be refused before it reaches the store for any cluster central
|
||||||
|
/// does not know — which is the Phase 1A known-cluster-set gate, where central learns its
|
||||||
|
/// cluster set from the DB. Until then, uniform text buys content-level indistinguishability
|
||||||
|
/// only.
|
||||||
/// </para>
|
/// </para>
|
||||||
/// <para>
|
/// <para>
|
||||||
/// <b>Exactly one public constructor.</b> <c>Grpc.AspNetCore</c> silently stops invoking an
|
/// <b>Exactly one public constructor.</b> <c>Grpc.AspNetCore</c> silently stops invoking an
|
||||||
@@ -58,9 +67,7 @@ namespace ZB.MOM.WW.OtOpcUa.Host.Configuration;
|
|||||||
public sealed class CentralCommandAuthInterceptor : Interceptor
|
public sealed class CentralCommandAuthInterceptor : Interceptor
|
||||||
{
|
{
|
||||||
private const string ServicePrefix = "/mesh_transport.v1.CentralCommandService/";
|
private const string ServicePrefix = "/mesh_transport.v1.CentralCommandService/";
|
||||||
private const string AuthorizationHeader = "authorization";
|
|
||||||
private const string ClusterHeader = "x-otopcua-cluster";
|
private const string ClusterHeader = "x-otopcua-cluster";
|
||||||
private const string BearerPrefix = "Bearer ";
|
|
||||||
|
|
||||||
private readonly IMeshPskProvider _pskProvider;
|
private readonly IMeshPskProvider _pskProvider;
|
||||||
private readonly ILogger<CentralCommandAuthInterceptor> _logger;
|
private readonly ILogger<CentralCommandAuthInterceptor> _logger;
|
||||||
@@ -131,7 +138,7 @@ public sealed class CentralCommandAuthInterceptor : Interceptor
|
|||||||
if (!context.Method.StartsWith(ServicePrefix, StringComparison.Ordinal))
|
if (!context.Method.StartsWith(ServicePrefix, StringComparison.Ordinal))
|
||||||
return;
|
return;
|
||||||
|
|
||||||
var clusterId = ExtractHeader(context.RequestHeaders, ClusterHeader);
|
var clusterId = GrpcRequestHeaders.First(context.RequestHeaders, ClusterHeader);
|
||||||
if (string.IsNullOrWhiteSpace(clusterId))
|
if (string.IsNullOrWhiteSpace(clusterId))
|
||||||
{
|
{
|
||||||
_logger.LogWarning(
|
_logger.LogWarning(
|
||||||
@@ -158,7 +165,7 @@ public sealed class CentralCommandAuthInterceptor : Interceptor
|
|||||||
"Central command authentication failed."));
|
"Central command authentication failed."));
|
||||||
}
|
}
|
||||||
|
|
||||||
var presented = ExtractBearerToken(context.RequestHeaders);
|
var presented = GrpcRequestHeaders.BearerToken(context.RequestHeaders);
|
||||||
if (presented is null || !FixedTimeEquals(presented, expected))
|
if (presented is null || !FixedTimeEquals(presented, expected))
|
||||||
{
|
{
|
||||||
_logger.LogWarning(
|
_logger.LogWarning(
|
||||||
@@ -172,27 +179,6 @@ public sealed class CentralCommandAuthInterceptor : Interceptor
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
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)
|
private static bool FixedTimeEquals(string presented, string expected)
|
||||||
=> CryptographicOperations.FixedTimeEquals(
|
=> CryptographicOperations.FixedTimeEquals(
|
||||||
Encoding.UTF8.GetBytes(presented), Encoding.UTF8.GetBytes(expected));
|
Encoding.UTF8.GetBytes(presented), Encoding.UTF8.GetBytes(expected));
|
||||||
|
|||||||
@@ -0,0 +1,70 @@
|
|||||||
|
using Grpc.Core;
|
||||||
|
|
||||||
|
namespace ZB.MOM.WW.OtOpcUa.Host.Configuration;
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// Request-metadata reads shared by the two mesh-command auth interceptors
|
||||||
|
/// (<see cref="MeshCommandAuthInterceptor"/>, <see cref="CentralCommandAuthInterceptor"/>).
|
||||||
|
/// </summary>
|
||||||
|
/// <remarks>
|
||||||
|
/// <para>
|
||||||
|
/// Exists because those two are documented structural clones of each other, and the one place
|
||||||
|
/// a clone pair can silently diverge is header parsing — where a difference is not a style
|
||||||
|
/// difference but a different answer to "which of these two <c>authorization</c> headers
|
||||||
|
/// counts". They had exactly that divergence before this type: one took the first entry, the
|
||||||
|
/// other scanned for the first <c>Bearer</c>-prefixed entry, so a request carrying
|
||||||
|
/// <c>authorization: junk</c> followed by <c>authorization: Bearer <psk></c> was denied
|
||||||
|
/// by one and accepted by the other.
|
||||||
|
/// </para>
|
||||||
|
/// <para>
|
||||||
|
/// <b>First entry wins, and it must parse.</b> A non-<c>Bearer</c> first entry is a failed
|
||||||
|
/// authentication, never a reason to keep looking: scanning past it lets a caller smuggle a
|
||||||
|
/// credential behind a header some intermediary may have inspected, matched or rewritten, and
|
||||||
|
/// makes which credential is checked depend on ordering nothing guarantees.
|
||||||
|
/// </para>
|
||||||
|
/// <para>
|
||||||
|
/// The three pre-existing sibling interceptors (LocalDb sync, config-serve, telemetry) keep
|
||||||
|
/// their own private copies deliberately — they are a shipped, live-gated wire contract, and
|
||||||
|
/// tightening their header handling is a behaviour change to three running endpoints, not a
|
||||||
|
/// refactor.
|
||||||
|
/// </para>
|
||||||
|
/// </remarks>
|
||||||
|
internal static class GrpcRequestHeaders
|
||||||
|
{
|
||||||
|
/// <summary>The bearer-token header, lowercased as gRPC puts it on the wire.</summary>
|
||||||
|
internal const string Authorization = "authorization";
|
||||||
|
|
||||||
|
/// <summary>Scheme prefix expected on the <see cref="Authorization"/> value.</summary>
|
||||||
|
internal const string BearerPrefix = "Bearer ";
|
||||||
|
|
||||||
|
/// <summary>Gets the first value carried under <paramref name="name"/>, or null if absent.</summary>
|
||||||
|
/// <param name="headers">The call's request metadata.</param>
|
||||||
|
/// <param name="name">The header name, matched case-insensitively.</param>
|
||||||
|
/// <returns>The first matching value, or <c>null</c>.</returns>
|
||||||
|
internal static string? First(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;
|
||||||
|
}
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// Gets the token from the first <c>authorization</c> header, or null when that header is
|
||||||
|
/// absent or does not carry the <c>Bearer</c> scheme.
|
||||||
|
/// </summary>
|
||||||
|
/// <param name="headers">The call's request metadata.</param>
|
||||||
|
/// <returns>The bearer token, or <c>null</c>.</returns>
|
||||||
|
internal static string? BearerToken(Metadata headers)
|
||||||
|
{
|
||||||
|
var value = First(headers, Authorization);
|
||||||
|
return value is not null && value.StartsWith(BearerPrefix, StringComparison.OrdinalIgnoreCase)
|
||||||
|
? value[BearerPrefix.Length..]
|
||||||
|
: null;
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -43,6 +43,12 @@ namespace ZB.MOM.WW.OtOpcUa.Host.Configuration;
|
|||||||
/// unavoidably observable and are not sensitive.
|
/// unavoidably observable and are not sensitive.
|
||||||
/// </para>
|
/// </para>
|
||||||
/// <para>
|
/// <para>
|
||||||
|
/// Header reads go through <see cref="GrpcRequestHeaders"/>, shared with
|
||||||
|
/// <see cref="CentralCommandAuthInterceptor"/> — the two are clones, and header parsing is
|
||||||
|
/// the one place a clone pair can diverge into two different answers about which of several
|
||||||
|
/// <c>authorization</c> headers counts.
|
||||||
|
/// </para>
|
||||||
|
/// <para>
|
||||||
/// <b>Exactly one public constructor.</b> <c>Grpc.AspNetCore</c> silently stops invoking an
|
/// <b>Exactly one public constructor.</b> <c>Grpc.AspNetCore</c> silently stops invoking an
|
||||||
/// interceptor that has more than one public ctor — see
|
/// interceptor that has more than one public ctor — see
|
||||||
/// <c>MeshCommandAuthInterceptorTests.HasExactlyOnePublicConstructor</c>.
|
/// <c>MeshCommandAuthInterceptorTests.HasExactlyOnePublicConstructor</c>.
|
||||||
@@ -51,8 +57,6 @@ namespace ZB.MOM.WW.OtOpcUa.Host.Configuration;
|
|||||||
public sealed class MeshCommandAuthInterceptor : Interceptor
|
public sealed class MeshCommandAuthInterceptor : Interceptor
|
||||||
{
|
{
|
||||||
private const string ServicePrefix = "/mesh_transport.v1.NodeCommandService/";
|
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 IOptions<MeshTransportOptions> _options;
|
||||||
private readonly ILogger<MeshCommandAuthInterceptor> _logger;
|
private readonly ILogger<MeshCommandAuthInterceptor> _logger;
|
||||||
@@ -137,7 +141,7 @@ public sealed class MeshCommandAuthInterceptor : Interceptor
|
|||||||
"Node command service is not accepting calls: no pre-shared key is configured on this node."));
|
"Node command service is not accepting calls: no pre-shared key is configured on this node."));
|
||||||
}
|
}
|
||||||
|
|
||||||
var presented = ExtractBearerToken(context.RequestHeaders);
|
var presented = GrpcRequestHeaders.BearerToken(context.RequestHeaders);
|
||||||
if (presented is null || !FixedTimeEquals(presented, expected))
|
if (presented is null || !FixedTimeEquals(presented, expected))
|
||||||
{
|
{
|
||||||
_logger.LogWarning(
|
_logger.LogWarning(
|
||||||
@@ -150,23 +154,6 @@ public sealed class MeshCommandAuthInterceptor : Interceptor
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
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)
|
private static bool FixedTimeEquals(string presented, string expected)
|
||||||
=> CryptographicOperations.FixedTimeEquals(
|
=> CryptographicOperations.FixedTimeEquals(
|
||||||
Encoding.UTF8.GetBytes(presented), Encoding.UTF8.GetBytes(expected));
|
Encoding.UTF8.GetBytes(presented), Encoding.UTF8.GetBytes(expected));
|
||||||
|
|||||||
+107
-11
@@ -1,3 +1,5 @@
|
|||||||
|
using Microsoft.Extensions.Configuration;
|
||||||
|
using Microsoft.Extensions.Logging;
|
||||||
using Microsoft.Extensions.Logging.Abstractions;
|
using Microsoft.Extensions.Logging.Abstractions;
|
||||||
using Microsoft.Extensions.Options;
|
using Microsoft.Extensions.Options;
|
||||||
using Shouldly;
|
using Shouldly;
|
||||||
@@ -22,7 +24,8 @@ public sealed class MeshPskProviderTests
|
|||||||
{
|
{
|
||||||
private static MeshPskProvider CreateProvider(
|
private static MeshPskProvider CreateProvider(
|
||||||
ISecretResolver resolver,
|
ISecretResolver resolver,
|
||||||
Dictionary<string, string>? clusterPsks = null)
|
Dictionary<string, string>? clusterPsks = null,
|
||||||
|
ILogger<MeshPskProvider>? logger = null)
|
||||||
{
|
{
|
||||||
var options = new MeshTransportOptions();
|
var options = new MeshTransportOptions();
|
||||||
if (clusterPsks is not null)
|
if (clusterPsks is not null)
|
||||||
@@ -32,7 +35,7 @@ public sealed class MeshPskProviderTests
|
|||||||
}
|
}
|
||||||
|
|
||||||
return new MeshPskProvider(
|
return new MeshPskProvider(
|
||||||
Options.Create(options), resolver, NullLogger<MeshPskProvider>.Instance);
|
Options.Create(options), resolver, logger ?? NullLogger<MeshPskProvider>.Instance);
|
||||||
}
|
}
|
||||||
|
|
||||||
[Fact]
|
[Fact]
|
||||||
@@ -275,28 +278,92 @@ public sealed class MeshPskProviderTests
|
|||||||
resolver.CallCount.ShouldBe(0);
|
resolver.CallCount.ShouldBe(0);
|
||||||
}
|
}
|
||||||
|
|
||||||
[Fact]
|
[Fact(Timeout = 30_000)]
|
||||||
public async Task ConcurrentCallers_ResolveOnce_AndAllSeeTheSameValue()
|
public async Task ConcurrentCallers_ResolveOnce_AndAllSeeTheSameValue()
|
||||||
{
|
{
|
||||||
// The gRPC gate calls this from every inbound ack, on arbitrary thread-pool threads. A
|
// The gRPC gate calls this from every inbound ack, on arbitrary thread-pool threads. A
|
||||||
// non-atomic fill would stampede the store and could hand two callers different keys.
|
// non-atomic fill would stampede the store and could hand two callers different keys.
|
||||||
var gate = new SemaphoreSlim(0, 1);
|
//
|
||||||
var resolver = new FakeSecretResolver { Value = "stored-psk", Gate = gate };
|
// The shape matters more than the assertion. Callers park BEFORE TryGetPsk and are released
|
||||||
|
// together, so the race is real rather than accidentally serialized by task scheduling —
|
||||||
|
// parking them INSIDE the resolver could never work, because the whole point of the Lazy is
|
||||||
|
// that only one caller ever gets in there. And the resolver's delay RETURNS rather than
|
||||||
|
// waiting on a gate: a regression that lets a second caller reach the store must drive
|
||||||
|
// CallCount to 2 (red), never sit on an exhausted semaphore (hang, which reads as a flake).
|
||||||
|
const int Callers = 16;
|
||||||
|
using var arrived = new CountdownEvent(Callers);
|
||||||
|
using var start = new ManualResetEventSlim(false);
|
||||||
|
var resolver = new FakeSecretResolver { Value = "stored-psk", ResolveDelay = TimeSpan.FromMilliseconds(50) };
|
||||||
var provider = CreateProvider(resolver);
|
var provider = CreateProvider(resolver);
|
||||||
|
|
||||||
var callers = Enumerable.Range(0, 16).Select(_ => Task.Run(() =>
|
var callers = Enumerable.Range(0, Callers).Select(_ => Task.Run(() =>
|
||||||
{
|
{
|
||||||
|
arrived.Signal();
|
||||||
|
start.Wait();
|
||||||
provider.TryGetPsk("SITE-A", out var psk).ShouldBeTrue();
|
provider.TryGetPsk("SITE-A", out var psk).ShouldBeTrue();
|
||||||
return psk;
|
return psk;
|
||||||
})).ToArray();
|
})).ToArray();
|
||||||
|
|
||||||
gate.Release();
|
arrived.Wait(TimeSpan.FromSeconds(10)).ShouldBeTrue("all callers should reach the start gate");
|
||||||
|
start.Set();
|
||||||
var results = await Task.WhenAll(callers);
|
var results = await Task.WhenAll(callers);
|
||||||
|
|
||||||
results.ShouldAllBe(r => r == "stored-psk");
|
results.ShouldAllBe(r => r == "stored-psk");
|
||||||
resolver.CallCount.ShouldBe(1);
|
resolver.CallCount.ShouldBe(1);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
[Fact]
|
||||||
|
public void APoisonedCacheEntry_IsEvicted_RatherThanDenyingTheClusterForever()
|
||||||
|
{
|
||||||
|
// ResolveFromStore swallows what the store throws, but it LOGS while doing so — and a logger
|
||||||
|
// provider that throws escapes the factory. Lazy under ExecutionAndPublication caches that
|
||||||
|
// exception and rethrows it on every later read, so an unguarded entry.Value would deny this
|
||||||
|
// cluster for the life of the process; nothing calls Invalidate until Phase 1A.
|
||||||
|
var logger = new ThrowingLogger { ThrowOnLog = true };
|
||||||
|
var resolver = new FakeSecretResolver { Throw = new InvalidOperationException("KEK unavailable") };
|
||||||
|
var provider = CreateProvider(resolver, logger: logger);
|
||||||
|
|
||||||
|
// The logger throwing must not escape the gate either — a denial, not a 500.
|
||||||
|
provider.TryGetPsk("SITE-A", out var first).ShouldBeFalse();
|
||||||
|
first.ShouldBeNull();
|
||||||
|
resolver.CallCount.ShouldBe(1);
|
||||||
|
|
||||||
|
logger.ThrowOnLog = false;
|
||||||
|
resolver.Throw = null;
|
||||||
|
resolver.Value = "stored-psk";
|
||||||
|
|
||||||
|
provider.TryGetPsk("SITE-A", out var second).ShouldBeTrue();
|
||||||
|
second.ShouldBe("stored-psk");
|
||||||
|
resolver.CallCount.ShouldBe(2); // reached the store again — the entry was not poisoned
|
||||||
|
}
|
||||||
|
|
||||||
|
[Fact]
|
||||||
|
public void ConfigMapLookup_StaysCaseInsensitive_ThroughRealConfigurationBinding()
|
||||||
|
{
|
||||||
|
// Writing into the options object pins only the field initializer's comparer. What reaches
|
||||||
|
// production is a BOUND instance, and the binder is free to construct its own Dictionary and
|
||||||
|
// drop the comparer silently — the symptom being a rig lookup that misses on a ClusterId
|
||||||
|
// spelled the way the DB spells it. Bind for real, then go through the PROVIDER.
|
||||||
|
var configuration = new ConfigurationBuilder()
|
||||||
|
.AddInMemoryCollection(new Dictionary<string, string?>
|
||||||
|
{
|
||||||
|
["MeshTransport:ClusterPsks:SITE-A"] = "rig-psk-site-a",
|
||||||
|
})
|
||||||
|
.Build();
|
||||||
|
|
||||||
|
var options = new MeshTransportOptions();
|
||||||
|
configuration.GetSection(MeshTransportOptions.SectionName).Bind(options);
|
||||||
|
|
||||||
|
var resolver = new FakeSecretResolver();
|
||||||
|
var provider = new MeshPskProvider(
|
||||||
|
Options.Create(options), resolver, NullLogger<MeshPskProvider>.Instance);
|
||||||
|
|
||||||
|
provider.TryGetPsk("site-a", out var psk).ShouldBeTrue();
|
||||||
|
|
||||||
|
psk.ShouldBe("rig-psk-site-a");
|
||||||
|
resolver.CallCount.ShouldBe(0);
|
||||||
|
}
|
||||||
|
|
||||||
/// <summary>
|
/// <summary>
|
||||||
/// An <see cref="ISecretResolver"/> whose answer the test controls between calls, so cache
|
/// An <see cref="ISecretResolver"/> whose answer the test controls between calls, so cache
|
||||||
/// staleness and invalidation are observable without a real store.
|
/// staleness and invalidation are observable without a real store.
|
||||||
@@ -311,8 +378,12 @@ public sealed class MeshPskProviderTests
|
|||||||
/// <summary>When set, <see cref="GetAsync"/> throws it instead of returning.</summary>
|
/// <summary>When set, <see cref="GetAsync"/> throws it instead of returning.</summary>
|
||||||
public Exception? Throw { get; set; }
|
public Exception? Throw { get; set; }
|
||||||
|
|
||||||
/// <summary>When set, the first call waits on it — used to force a concurrent stampede.</summary>
|
/// <summary>
|
||||||
public SemaphoreSlim? Gate { get; set; }
|
/// How long each call takes before answering. Widens the window a concurrent stampede
|
||||||
|
/// would have to exploit — and it RETURNS, so a second entrant shows up as a second call
|
||||||
|
/// rather than as a hang.
|
||||||
|
/// </summary>
|
||||||
|
public TimeSpan ResolveDelay { get; set; } = TimeSpan.Zero;
|
||||||
|
|
||||||
/// <summary>How many times the store was consulted.</summary>
|
/// <summary>How many times the store was consulted.</summary>
|
||||||
public int CallCount => Volatile.Read(ref _callCount);
|
public int CallCount => Volatile.Read(ref _callCount);
|
||||||
@@ -325,8 +396,8 @@ public sealed class MeshPskProviderTests
|
|||||||
LastName = name.Value;
|
LastName = name.Value;
|
||||||
Interlocked.Increment(ref _callCount);
|
Interlocked.Increment(ref _callCount);
|
||||||
|
|
||||||
if (Gate is not null)
|
if (ResolveDelay > TimeSpan.Zero)
|
||||||
await Gate.WaitAsync(ct).ConfigureAwait(false);
|
await Task.Delay(ResolveDelay, ct).ConfigureAwait(false);
|
||||||
|
|
||||||
if (Throw is not null)
|
if (Throw is not null)
|
||||||
throw Throw;
|
throw Throw;
|
||||||
@@ -334,4 +405,29 @@ public sealed class MeshPskProviderTests
|
|||||||
return Value;
|
return Value;
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// An <see cref="ILogger{T}"/> that can be made to throw, standing in for a logger provider
|
||||||
|
/// that faults — the only way the provider's own catch blocks can throw.
|
||||||
|
/// </summary>
|
||||||
|
private sealed class ThrowingLogger : ILogger<MeshPskProvider>
|
||||||
|
{
|
||||||
|
/// <summary>When true, every <see cref="Log"/> call throws.</summary>
|
||||||
|
public bool ThrowOnLog { get; set; }
|
||||||
|
|
||||||
|
public IDisposable? BeginScope<TState>(TState state) where TState : notnull => null;
|
||||||
|
|
||||||
|
public bool IsEnabled(LogLevel logLevel) => true;
|
||||||
|
|
||||||
|
public void Log<TState>(
|
||||||
|
LogLevel logLevel,
|
||||||
|
EventId eventId,
|
||||||
|
TState state,
|
||||||
|
Exception? exception,
|
||||||
|
Func<TState, Exception?, string> formatter)
|
||||||
|
{
|
||||||
|
if (ThrowOnLog)
|
||||||
|
throw new InvalidOperationException("logger provider faulted");
|
||||||
|
}
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
+50
-2
@@ -30,10 +30,17 @@ public sealed class CentralCommandAuthInterceptorTests
|
|||||||
|
|
||||||
private static ServerCallContext CreateContext(
|
private static ServerCallContext CreateContext(
|
||||||
string method, string? authorizationHeader, string? clusterHeader)
|
string method, string? authorizationHeader, string? clusterHeader)
|
||||||
|
=> CreateContext(
|
||||||
|
method,
|
||||||
|
authorizationHeader is null ? [] : [authorizationHeader],
|
||||||
|
clusterHeader);
|
||||||
|
|
||||||
|
private static ServerCallContext CreateContext(
|
||||||
|
string method, string[] authorizationHeaders, string? clusterHeader)
|
||||||
{
|
{
|
||||||
var headers = new Metadata();
|
var headers = new Metadata();
|
||||||
if (authorizationHeader is not null)
|
foreach (var value in authorizationHeaders)
|
||||||
headers.Add("authorization", authorizationHeader);
|
headers.Add("authorization", value);
|
||||||
if (clusterHeader is not null)
|
if (clusterHeader is not null)
|
||||||
headers.Add("x-otopcua-cluster", clusterHeader);
|
headers.Add("x-otopcua-cluster", clusterHeader);
|
||||||
|
|
||||||
@@ -143,6 +150,47 @@ public sealed class CentralCommandAuthInterceptorTests
|
|||||||
ex.StatusCode.ShouldBe(StatusCode.PermissionDenied);
|
ex.StatusCode.ShouldBe(StatusCode.PermissionDenied);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
[Fact]
|
||||||
|
public async Task ReportApply_WithAJunkAuthorizationHeaderBeforeTheRealOne_IsDenied()
|
||||||
|
{
|
||||||
|
// First entry wins, and it must parse. Scanning past a non-Bearer header for a later one
|
||||||
|
// would let a caller smuggle its credential behind a header an intermediary may have
|
||||||
|
// inspected or rewritten, and would make WHICH credential is checked depend on ordering
|
||||||
|
// nothing guarantees. Pinned identically on MeshCommandAuthInterceptor.
|
||||||
|
var interceptor = CreateInterceptor(("SITE-A", "psk-site-a"));
|
||||||
|
var context = CreateContext(
|
||||||
|
ReportApplyMethod, ["junk", "Bearer psk-site-a"], "SITE-A");
|
||||||
|
|
||||||
|
var ex = await Should.ThrowAsync<RpcException>(() => Invoke(interceptor, context));
|
||||||
|
ex.StatusCode.ShouldBe(StatusCode.PermissionDenied);
|
||||||
|
}
|
||||||
|
|
||||||
|
[Fact]
|
||||||
|
public async Task EveryDenialArm_ReturnsAnIdenticalDetail()
|
||||||
|
{
|
||||||
|
// The anti-enumeration mechanism IS this identity: the caller supplies the cluster id, so a
|
||||||
|
// reply distinguishing "no key for that cluster" from "wrong key" would let an
|
||||||
|
// unauthenticated caller read central's cluster set off the wire. Asserting only StatusCode
|
||||||
|
// would let a well-meaning "more helpful message" edit break it silently.
|
||||||
|
// (Content-level only — the class remark records the timing oracle that survives until the
|
||||||
|
// Phase 1A known-cluster-set gate.)
|
||||||
|
var interceptor = CreateInterceptor(("SITE-A", "psk-site-a"));
|
||||||
|
|
||||||
|
var noHeader = await Should.ThrowAsync<RpcException>(() => Invoke(
|
||||||
|
interceptor, CreateContext(ReportApplyMethod, "Bearer psk-site-a", clusterHeader: null)));
|
||||||
|
var unknownCluster = await Should.ThrowAsync<RpcException>(() => Invoke(
|
||||||
|
interceptor, CreateContext(ReportApplyMethod, "Bearer psk-site-a", "SITE-Z")));
|
||||||
|
var wrongToken = await Should.ThrowAsync<RpcException>(() => Invoke(
|
||||||
|
interceptor, CreateContext(ReportApplyMethod, "Bearer wrong", "SITE-A")));
|
||||||
|
var noToken = await Should.ThrowAsync<RpcException>(() => Invoke(
|
||||||
|
interceptor, CreateContext(ReportApplyMethod, authorizationHeader: null, clusterHeader: "SITE-A")));
|
||||||
|
|
||||||
|
unknownCluster.Status.Detail.ShouldBe(noHeader.Status.Detail);
|
||||||
|
wrongToken.Status.Detail.ShouldBe(noHeader.Status.Detail);
|
||||||
|
noToken.Status.Detail.ShouldBe(noHeader.Status.Detail);
|
||||||
|
noHeader.Status.Detail.ShouldNotBeNullOrWhiteSpace();
|
||||||
|
}
|
||||||
|
|
||||||
[Fact]
|
[Fact]
|
||||||
public async Task ServerStreamingPath_IsAlsoGated()
|
public async Task ServerStreamingPath_IsAlsoGated()
|
||||||
{
|
{
|
||||||
|
|||||||
+19
-2
@@ -31,10 +31,13 @@ public sealed class MeshCommandAuthInterceptorTests
|
|||||||
NullLogger<MeshCommandAuthInterceptor>.Instance);
|
NullLogger<MeshCommandAuthInterceptor>.Instance);
|
||||||
|
|
||||||
private static ServerCallContext CreateContext(string method, string? authorizationHeader)
|
private static ServerCallContext CreateContext(string method, string? authorizationHeader)
|
||||||
|
=> CreateContext(method, authorizationHeader is null ? [] : [authorizationHeader]);
|
||||||
|
|
||||||
|
private static ServerCallContext CreateContext(string method, string[] authorizationHeaders)
|
||||||
{
|
{
|
||||||
var headers = new Metadata();
|
var headers = new Metadata();
|
||||||
if (authorizationHeader is not null)
|
foreach (var value in authorizationHeaders)
|
||||||
headers.Add("authorization", authorizationHeader);
|
headers.Add("authorization", value);
|
||||||
|
|
||||||
return new FakeServerCallContext(method, headers);
|
return new FakeServerCallContext(method, headers);
|
||||||
}
|
}
|
||||||
@@ -133,6 +136,20 @@ public sealed class MeshCommandAuthInterceptorTests
|
|||||||
await Invoke(interceptor, context); // no throw
|
await Invoke(interceptor, context); // no throw
|
||||||
}
|
}
|
||||||
|
|
||||||
|
[Fact]
|
||||||
|
public async Task ExecuteMethod_WithAJunkAuthorizationHeaderBeforeTheRealOne_IsDenied()
|
||||||
|
{
|
||||||
|
// First entry wins, and it must parse. This interceptor originally scanned past a non-Bearer
|
||||||
|
// header for a later Bearer one while its central-side clone took the first entry, so the
|
||||||
|
// two clones gave OPPOSITE answers to this exact request. Both now share
|
||||||
|
// GrpcRequestHeaders and both deny it; pinned identically on CentralCommandAuthInterceptor.
|
||||||
|
var interceptor = CreateInterceptor("the-cluster-psk");
|
||||||
|
var context = CreateContext(ExecuteMethod, ["junk", "Bearer the-cluster-psk"]);
|
||||||
|
|
||||||
|
var ex = await Should.ThrowAsync<RpcException>(() => Invoke(interceptor, context));
|
||||||
|
ex.StatusCode.ShouldBe(StatusCode.PermissionDenied);
|
||||||
|
}
|
||||||
|
|
||||||
[Fact]
|
[Fact]
|
||||||
public async Task ExecuteMethod_TokenComparison_IsNotAPrefixMatch()
|
public async Task ExecuteMethod_TokenComparison_IsNotAPrefixMatch()
|
||||||
{
|
{
|
||||||
|
|||||||
Reference in New Issue
Block a user