fix(host): follow an in-place site PSK rotation without a central restart (ScadaBridge#96)

SitePskProvider cached each site's gRPC preshared key forever; only site removal invalidated it, so rotating SB-GRPC-PSK-{site} broke the central<->site link on both central nodes until they restarted. The cached key is now dropped when: - the secret is written through /admin/secrets: central replaces the secrets library's ISecretCacheInvalidator with SitePskRotationInvalidator, which also publishes SitePskInvalidated on a new DistributedPubSub topic (site-psk-invalidation) so the peer central node drops its copy and its resolver cache entry too (SitePskInvalidationSubscriber); - a site rejects a central->site call with PermissionDenied/Unauthenticated: WithSiteCredentials wraps every per-site channel's handler in PskRejectionRefreshHandler, so the next call re-reads the key (covers out-of-process writes such as the secret CLI); - a site->central call presents a token that does not match the cached key: CentralControlAuthInterceptor invalidates and re-reads once before refusing; - ScadaBridge:Communication:SitePsks reloads. Invalidation is case-insensitive because SecretName lowercases names. Rotation runbook (topology-guide), Component-Communication/Host and the #85 cache audit (D2) updated. Claude-Session: https://claude.ai/code/session_01GT67AEHd2Acv1YPwGfXEoV
2026-11-09 14:27:14 -05:00
parent 7f479f3a5c
commit 3a31dd2de7
15 changed files with 666 additions and 26 deletions
@@ -274,4 +274,58 @@ public class CentralControlAuthInterceptorTests
            CentralControlAuthInterceptor.DefaultGatedPrefixes,
            p => p == $"/{method}/");
    }
+    /// <summary>A provider whose cached key is stale until <see cref="Invalidate"/> runs.</summary>
+    private sealed class RotatingPskProvider(string cached, string rotated) : ISitePskProvider
+    {
+        private string _current = cached;
+
+        /// <summary>Gets the number of invalidations.</summary>
+        public int Invalidations { get; private set; }
+
+        /// <inheritdoc />
+        public ValueTask<string> GetAsync(string siteId, CancellationToken ct) => new(_current);
+
+        /// <inheritdoc />
+        public void Invalidate(string siteId)
+        {
+            Invalidations++;
+            _current = rotated;
+        }
+    }
+
+    /// <summary>
+    /// Gitea #96: a site already on its rotated key is accepted even while central's cached copy
+    /// is stale — the mismatch drops the cache and re-reads once.
+    /// </summary>
+    /// <returns>A task that represents the asynchronous operation.</returns>
+    [Fact]
+    public async Task StaleCachedKey_IsReReadOnce_SoARotatedSiteIsAccepted()
+    {
+        var provider = new RotatingPskProvider("pre-rotation", "rotated");
+        var interceptor = new CentralControlAuthInterceptor(
+            provider, NullLogger<CentralControlAuthInterceptor>.Instance);
+
+        var result = await Invoke(interceptor, CreateContext(ControlMethod, SiteA, "Bearer rotated"));
+
+        Assert.Equal("ok", result);
+        Assert.Equal(1, provider.Invalidations);
+    }
+
+    /// <summary>Negative control: a wrong token is still refused after the one re-read, and a right one never invalidates.</summary>
+    /// <returns>A task that represents the asynchronous operation.</returns>
+    [Fact]
+    public async Task WrongToken_IsStillRefused_AfterTheReRead_AndAMatchNeverInvalidates()
+    {
+        var provider = new RotatingPskProvider("k", "k");
+        var interceptor = new CentralControlAuthInterceptor(
+            provider, NullLogger<CentralControlAuthInterceptor>.Instance);
+
+        Assert.Equal("ok", await Invoke(interceptor, CreateContext(ControlMethod, SiteA, "Bearer k")));
+        Assert.Equal(0, provider.Invalidations);
+
+        var ex = await Assert.ThrowsAsync<RpcException>(
+            () => Invoke(interceptor, CreateContext(ControlMethod, SiteA, "Bearer wrong")));
+        Assert.Equal(StatusCode.PermissionDenied, ex.StatusCode);
+        Assert.Equal(1, provider.Invalidations);
+    }
 }
@@ -407,6 +407,18 @@ public class CentralCompositionRootTests : IDisposable
        Assert.Null(_factory.Services.GetService<IInboundApiKeyStoreImporter>());
    }

+    /// <summary>
+    /// Gitea #96: the secrets write path's <see cref="ZB.MOM.WW.Secrets.Abstractions.ISecretCacheInvalidator"/>
+    /// must be the PSK-aware wrapper on central, or an in-place site-PSK rotation through
+    /// /admin/secrets never reaches the PSK cache.
+    /// </summary>
+    [Fact]
+    public void Central_SecretCacheInvalidator_IsTheSitePskRotationInvalidator()
+    {
+        Assert.IsType<SitePskRotationInvalidator>(
+            _factory.Services.GetService<ZB.MOM.WW.Secrets.Abstractions.ISecretCacheInvalidator>());
+    }
+
    /// <summary>Verifies that the shared audit seam resolves to the central direct-write bridge.</summary>
    [Fact]
    public void Central_SharedAuditSeam_IsTheCentralDirectWriteBridge()
@@ -7,8 +7,10 @@ using Microsoft.Extensions.DependencyInjection;
 using Microsoft.Extensions.Hosting;
 using Microsoft.Extensions.Logging.Abstractions;
 using Microsoft.Extensions.Options;
+using NSubstitute;
 using ZB.MOM.WW.ScadaBridge.Communication;
 using ZB.MOM.WW.ScadaBridge.Communication.Grpc;
+using ZB.MOM.WW.Secrets.Abstractions;

 namespace ZB.MOM.WW.ScadaBridge.Host.Tests;

@@ -267,4 +269,87 @@ public class ControlPlaneAuthEndToEndTests : IAsyncLifetime
        var service = _host.Services.GetRequiredService<EchoSiteStreamService>();
        Assert.Equal("site-a", service.LastSiteHeader);
    }
+    // --- Gitea #96: a site's rejection drops central's cached key ---------------------------
+
+    /// <summary>
+    /// Central's real <see cref="SitePskProvider"/> over a store that served a stale key first and
+    /// the rotated (site-accepted) key after: the in-place rotation scenario.
+    /// </summary>
+    private static (SitePskProvider Provider, ISecretResolver Resolver) RotatedProvider()
+    {
+        var resolver = Substitute.For<ISecretResolver>();
+        resolver.GetAsync(Arg.Any<SecretName>(), Arg.Any<CancellationToken>())
+            .Returns("the-pre-rotation-key", SiteKey);
+        var options = Substitute.For<IOptionsMonitor<CommunicationOptions>>();
+        options.CurrentValue.Returns(new CommunicationOptions());
+        return (new SitePskProvider(resolver, options, NullLogger<SitePskProvider>.Instance), resolver);
+    }
+
+    private GrpcChannel Channel(ISitePskProvider provider) =>
+        GrpcChannel.ForAddress(_server.BaseAddress, new GrpcChannelOptions { HttpHandler = _server.CreateHandler() }
+            .WithMessageLimits().WithSiteCredentials(provider, "site-a"));
+
+    /// <summary>
+    /// Verifies that after a site rejects the cached (pre-rotation) key on a unary call, the next
+    /// call re-reads the key and is accepted — no restart needed.
+    /// </summary>
+    /// <returns>A task that represents the asynchronous operation.</returns>
+    [Fact]
+    public async Task RejectedUnaryCall_DropsTheCachedKey_SoTheNextCallUsesTheRotatedOne()
+    {
+        var (provider, _) = RotatedProvider();
+        using var channel = Channel(provider);
+        var client = new SiteStreamService.SiteStreamServiceClient(channel);
+
+        var ex = await Assert.ThrowsAsync<RpcException>(
+            async () => await client.PullAuditEventsAsync(new PullAuditEventsRequest()));
+        Assert.Equal(StatusCode.PermissionDenied, ex.StatusCode);
+
+        var reply = await client.PullAuditEventsAsync(new PullAuditEventsRequest { BatchSize = 1 });
+        Assert.False(reply.MoreAvailable);
+    }
+
+    /// <summary>Verifies the same refresh for a rejected streaming call.</summary>
+    /// <returns>A task that represents the asynchronous operation.</returns>
+    [Fact]
+    public async Task RejectedStreamingCall_DropsTheCachedKey_SoTheNextCallUsesTheRotatedOne()
+    {
+        var (provider, _) = RotatedProvider();
+        using var channel = Channel(provider);
+        var client = new SiteStreamService.SiteStreamServiceClient(channel);
+
+        using (var rejected = client.SubscribeInstance(new InstanceStreamRequest { CorrelationId = "c1" }))
+        {
+            var ex = await Assert.ThrowsAsync<RpcException>(
+                async () => await rejected.ResponseStream.MoveNext(CancellationToken.None));
+            Assert.Equal(StatusCode.PermissionDenied, ex.StatusCode);
+        }
+
+        using var accepted = client.SubscribeInstance(new InstanceStreamRequest { CorrelationId = "c2" });
+        Assert.True(await accepted.ResponseStream.MoveNext(CancellationToken.None));
+        Assert.Equal("c2", accepted.ResponseStream.Current.CorrelationId);
+    }
+
+    /// <summary>
+    /// Negative control: accepted calls never drop the cache, so the store is read once no
+    /// matter how many calls succeed.
+    /// </summary>
+    /// <returns>A task that represents the asynchronous operation.</returns>
+    [Fact]
+    public async Task AcceptedCalls_KeepTheCachedKey()
+    {
+        var resolver = Substitute.For<ISecretResolver>();
+        resolver.GetAsync(Arg.Any<SecretName>(), Arg.Any<CancellationToken>()).Returns(SiteKey);
+        var options = Substitute.For<IOptionsMonitor<CommunicationOptions>>();
+        options.CurrentValue.Returns(new CommunicationOptions());
+        var provider = new SitePskProvider(resolver, options, NullLogger<SitePskProvider>.Instance);
+        using var channel = Channel(provider);
+        var client = new SiteStreamService.SiteStreamServiceClient(channel);
+
+        await client.PullAuditEventsAsync(new PullAuditEventsRequest());
+        await client.PullAuditEventsAsync(new PullAuditEventsRequest());
+        await client.PullAuditEventsAsync(new PullAuditEventsRequest());
+
+        await resolver.Received(1).GetAsync(Arg.Any<SecretName>(), Arg.Any<CancellationToken>());
+    }
 }
@@ -204,4 +204,61 @@ public class SitePskProviderTests
        Assert.Equal("key-a", await provider.GetAsync("site-a", CancellationToken.None));
        Assert.Equal("key-b", await provider.GetAsync("site-b", CancellationToken.None));
    }
+    /// <summary>
+    /// Verifies that a reload of the configured <c>SitePsks</c> map drops cached keys, so a key
+    /// rotated in configuration is used without a restart (Gitea #96).
+    /// </summary>
+    /// <returns>A task that represents the asynchronous operation.</returns>
+    [Fact]
+    public async Task ReloadedSitePsks_DropsTheCache_SoAKeyRotatedInConfigIsUsed()
+    {
+        var resolver = Substitute.For<ISecretResolver>();
+        var options = new CommunicationOptions();
+        options.SitePsks["site-a"] = "old";
+        Action<CommunicationOptions, string?>? listener = null;
+        var monitor = Substitute.For<IOptionsMonitor<CommunicationOptions>>();
+        monitor.CurrentValue.Returns(_ => options);
+        monitor.OnChange(Arg.Do<Action<CommunicationOptions, string?>>(l => listener = l));
+        using var provider = new SitePskProvider(resolver, monitor, NullLogger<SitePskProvider>.Instance);
+
+        Assert.Equal("old", await provider.GetAsync("site-a", CancellationToken.None));
+
+        options = new CommunicationOptions();
+        options.SitePsks["site-a"] = "rotated";
+        Assert.NotNull(listener);
+        listener!(options, null);
+
+        Assert.Equal("rotated", await provider.GetAsync("site-a", CancellationToken.None));
+    }
+
+    /// <summary>Verifies the secret-name to site-id mapping the rotation hook relies on.</summary>
+    /// <param name="name">A secret name.</param>
+    /// <param name="expected">The expected site id, or null.</param>
+    [Theory]
+    [InlineData("SB-GRPC-PSK-site-a", "site-a")]
+    [InlineData("SB-GRPC-PSK-", null)]
+    [InlineData("OTHER-SECRET", null)]
+    [InlineData("sb-grpc-psk-site-a", "site-a")]
+    public void SiteIdFromSecretName_MapsOnlyPskNames(string name, string? expected)
+    {
+        Assert.Equal(expected, SitePskProvider.SiteIdFromSecretName(name));
+    }
+    /// <summary>
+    /// Verifies that invalidation ignores case: the rotation hook derives the site id from a
+    /// lowercased secret name, but the site may be cached under its mixed-case identifier.
+    /// </summary>
+    /// <returns>A task that represents the asynchronous operation.</returns>
+    [Fact]
+    public async Task Invalidate_IgnoresCase_BecauseSecretNamesAreLowercased()
+    {
+        var resolver = Substitute.For<ISecretResolver>();
+        resolver.GetAsync(Arg.Any<SecretName>(), Arg.Any<CancellationToken>())
+            .Returns("old", "rotated");
+        var provider = Create(resolver);
+
+        Assert.Equal("old", await provider.GetAsync("Site-A", CancellationToken.None));
+        provider.Invalidate("site-a");
+
+        Assert.Equal("rotated", await provider.GetAsync("Site-A", CancellationToken.None));
+    }
 }
@@ -0,0 +1,89 @@
+using Akka.Actor;
+using Akka.Cluster.Tools.PublishSubscribe;
+using Akka.TestKit.Xunit2;
+using Microsoft.Extensions.Logging.Abstractions;
+using NSubstitute;
+using ZB.MOM.WW.ScadaBridge.Communication.Grpc;
+using ZB.MOM.WW.Secrets.Abstractions;
+
+namespace ZB.MOM.WW.ScadaBridge.Host.Tests;
+
+/// <summary>
+/// Gitea #96: an in-place rotation of <c>SB-GRPC-PSK-{siteId}</c> written through the secrets
+/// write path must drop central's cached key on this node and on the peer central node.
+/// </summary>
+public class SitePskRotationInvalidatorTests : TestKit
+{
+    private readonly ISecretCacheInvalidator _inner = Substitute.For<ISecretCacheInvalidator>();
+    private readonly ISitePskProvider _psk = Substitute.For<ISitePskProvider>();
+
+    private SitePskRotationInvalidator Create() =>
+        new(_inner, _psk, NullLogger<SitePskRotationInvalidator>.Instance);
+
+    /// <summary>
+    /// Verifies that writing a site's PSK secret evicts the resolver cache, drops the site's
+    /// cached key and publishes the invalidation to the peer.
+    /// </summary>
+    [Fact]
+    public void WritingAPskSecret_DropsTheCachedKey_AndTellsThePeer()
+    {
+        var mediator = CreateTestProbe();
+        var invalidator = Create();
+        invalidator.SetMediator(mediator.Ref);
+
+        // The secrets library lowercases names, so the site id arrives lowercased too.
+        invalidator.Invalidate(new SecretName("SB-GRPC-PSK-site-a"));
+
+        _inner.Received(1).Invalidate(new SecretName("SB-GRPC-PSK-site-a"));
+        _psk.Received(1).Invalidate("site-a");
+        var publish = mediator.ExpectMsg<Publish>();
+        Assert.Equal(SitePskRotationInvalidator.Topic, publish.Topic);
+        Assert.Equal(new SitePskInvalidated("site-a"), publish.Message);
+    }
+
+    /// <summary>Negative control: any other secret only evicts the resolver cache.</summary>
+    [Fact]
+    public void WritingAnyOtherSecret_OnlyEvictsTheResolverCache()
+    {
+        var mediator = CreateTestProbe();
+        var invalidator = Create();
+        invalidator.SetMediator(mediator.Ref);
+
+        invalidator.Invalidate(new SecretName("SMTP-PASSWORD"));
+
+        _inner.Received(1).Invalidate(new SecretName("SMTP-PASSWORD"));
+        _psk.DidNotReceiveWithAnyArgs().Invalidate(default!);
+        mediator.ExpectNoMsg(TimeSpan.FromMilliseconds(200));
+    }
+
+    /// <summary>Verifies that before the actor system exists an invalidation still applies locally.</summary>
+    [Fact]
+    public void WithoutAMediator_TheLocalInvalidationStillApplies()
+    {
+        Create().Invalidate(new SecretName("SB-GRPC-PSK-site-a"));
+
+        _psk.Received(1).Invalidate("site-a");
+    }
+
+    /// <summary>
+    /// Verifies that the subscriber applies a peer's invalidation to this node's resolver and PSK
+    /// caches, without re-publishing it.
+    /// </summary>
+    [Fact]
+    public void Subscriber_AppliesAPeerInvalidation_WithoutRepublishing()
+    {
+        var mediator = CreateTestProbe();
+        var invalidator = Create();
+        invalidator.SetMediator(mediator.Ref);
+        var subscriber = Sys.ActorOf(Props.Create(() => new SitePskInvalidationSubscriber(invalidator)));
+
+        subscriber.Tell(new SitePskInvalidated("site-b"));
+
+        AwaitAssert(() =>
+        {
+            _psk.Received(1).Invalidate("site-b");
+            _inner.Received(1).Invalidate(new SecretName("SB-GRPC-PSK-site-b"));
+        });
+        mediator.ExpectNoMsg(TimeSpan.FromMilliseconds(200));
+    }
+}