using System.Collections.Concurrent;
using System.Diagnostics.CodeAnalysis;
using System.Security.Cryptography;
using System.Text;
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. Phase 1A
/// wires them to CentralCommunicationActor's contact refresh, which is the point at
/// which central learns a cluster's membership (and so possibly its key) has changed — so a
/// rotated key is picked up within one ContactRefreshSeconds without a restart.
///
///
/// The known-cluster gate is a real security control, not an optimization (Phase 1A
/// security register item 1). publishes the cluster ids
/// central just read from ClusterNode; consults that set
/// first and refuses anything outside it in O(1), touching neither the config map nor the
/// secret store. Without it, an unauthenticated caller can pick an arbitrary
/// x-otopcua-cluster value and make central perform a secret-store read plus an audit
/// write per probe — and, because negatives are deliberately never cached, the cost
/// difference between a known and an unknown cluster is measurable from the wire, which
/// defeats the uniform denial text on the central interceptor.
///
///
/// Invalidate is not a revocation barrier. 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 Invalidate as
/// "stop serving this from memory", never as "this key is now dead everywhere".
///
///
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);
///
/// Publishes the set of application Cluster ids central currently knows about.
/// refuses anything outside it before consulting any source.
///
///
/// The known cluster ids, matched case-insensitively. An empty collection deactivates the
/// gate rather than refusing everything: central publishes this from its DB contact
/// refresh, and "the DB told me nothing" must not become a fleet-wide authentication outage.
/// Never having called this leaves the gate inactive for the same reason — the central
/// interceptor is live from the first request, before the first refresh has run.
///
///
/// Implementations must take a copy: the caller owns its collection and may reuse it. The
/// swap must also be atomic with respect to concurrent callers,
/// which run on arbitrary gRPC threads.
///
void SetKnownClusters(IReadOnlyCollection clusterIds);
///
/// Verifies a presented bearer token against one cluster's key, in constant time with
/// respect to the key's content.
///
/// The application Cluster the caller claims to be.
/// The bearer token the caller supplied.
/// true only when a key exists for that cluster and the token equals it.
///
///
/// This is the only comparison on the verify path (Phase 1A security register
/// item 2). Handing the raw key back so a caller can compare it is how ==
/// eventually appears in an interceptor and compiles; keeping the comparison behind this
/// member means the verify side never holds key material at all.
/// survives for the dial side, which has to send the key rather than compare it.
///
///
/// Declared without a default implementation on purpose: a security member a new
/// implementer can inherit silently is a member a new implementer never considers.
///
///
/// Length differences remain observable through timing; that is inherent to comparing
/// byte arrays of different lengths and is not sensitive. Content is not.
///
///
bool Matches(string clusterId, string presented);
/// 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-";
///
/// Ceiling on one store read. Not a tuning knob — a backstop: the read runs under the
/// publication lock, so a store that never answers would park the
/// calling gRPC thread and every other caller queued behind it for that cluster.
/// Expiry fails closed like any other fault.
///
private static readonly TimeSpan StoreReadTimeout = TimeSpan.FromSeconds(3);
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);
///
/// The known-cluster gate's membership set, or when the gate is
/// inactive (never published, or published empty — see ).
///
///
/// Swapped wholesale, never mutated in place, so a reader either sees the previous set or the
/// next one and never a half-built one. volatile is what makes the publish visible to
/// the gRPC threads doing the reading; no lock is needed because there is exactly one writer
/// (the contact-refresh handler, on the actor's own thread).
///
private volatile IReadOnlySet? _knownClusters;
/// 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;
}
// The known-cluster gate, ahead of BOTH sources. Deliberately silent: the caller that
// rejected the call already logs the denial, and a second line here would be a log flood
// driven by exactly the unauthenticated traffic this gate exists to make cheap.
//
// Ahead of the config map too, not just the store. That has a visible consequence — a
// MeshTransport:ClusterPsks entry for a cluster with no enabled, non-maintenance ClusterNode
// row stops resolving once central publishes its set — and it is the right one: central
// would neither dial such a cluster nor accept an ack from it, so serving a key for it only
// widens what an unauthenticated caller can probe.
var known = _knownClusters;
if (known is not null && !known.Contains(clusterId))
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);
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>(clusterId, entry));
return false;
}
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 bool Matches(string clusterId, string presented)
{
// Blank is refused on its own terms rather than by relying on the fact that a blank key
// never survives resolution: this is an authentication gate, and "both sides happened to be
// empty" must never be a pass.
if (string.IsNullOrWhiteSpace(presented))
return false;
if (!TryGetPsk(clusterId, out var expected))
return false;
return CryptographicOperations.FixedTimeEquals(
Encoding.UTF8.GetBytes(presented), Encoding.UTF8.GetBytes(expected));
}
///
public void SetKnownClusters(IReadOnlyCollection clusterIds)
{
ArgumentNullException.ThrowIfNull(clusterIds);
// Copied, so a caller reusing its list cannot widen the gate after the fact; empty collapses
// to null so the gate has exactly one "inactive" representation to test for.
_knownClusters = clusterIds.Count == 0
? null
: clusterIds.ToHashSet(StringComparer.OrdinalIgnoreCase);
}
///
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. What makes
/// the block acceptable is that it is bounded () and
/// 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.
///
///
/// 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;
}
using var cts = new CancellationTokenSource(StoreReadTimeout);
try
{
return _secretResolver.GetAsync(name, cts.Token).GetAwaiter().GetResult();
}
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(
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;
}
}
}