feat(secrets): Grpc hub client — pull-only sweep service + options + DI

Client half of the central-hosted secrets hub: a site node follows central and never pushes. - `ISecretsHubReader` + `GrpcSecretsHubClient` — the read half of the hub as a follower sees it (manifest + bulk fetch, nothing else). Bearer credential and a per-call deadline are attached per call; responses convert through `WireToDomain`. Faults are deliberately not caught here, so an unreachable hub can never be mistaken for an empty one. - `WireToDomain` — strict, per-row proto -> domain parse mirroring the Akka replicator's `SecretRowDto`: a name the domain would reject, a content type naming no enum member, or a non-round-trip timestamp is refused at the boundary. Explicit proto3 presence keeps null distinguishable from empty. - `GrpcSecretSyncService` — `SqlServerSecretSyncService` with the push half removed. One `ReconcileAsync` per tick (pull-only by the reconciler's own contract, and batched, so a cold-start pull set cannot exceed gRPC's 4 MB cap), never throws out of `ExecuteAsync`, warns once and retries on the next interval. The local store is left completely undecorated — the sweep is its only hub-ward writer. - `SecretsGrpcHubClientOptions` — fail-closed `Validate()` naming the exact configuration key, including the section a caller actually bound. The bearer token is never echoed by any message. - `AddZbSecretsGrpcHubClient` — binds, eagerly validates, and registers the channel under a service KEY so it cannot lose a `TryAdd` race to an application's own `GrpcChannel` and dial the wrong endpoint with the hub credential attached. Tests 52/52 (red-first), including a source-scan guard pinning that no client file names `ComputePushSet`, `ReadLocalAsync`, or `ApplyReplicatedAsync`, and a reflection pin that the reader seam exposes exactly two read operations. Claude-Session: https://claude.ai/code/session_014WNM4vjoVksyyBraTXSZE1
2026-08-07 06:12:03 -04:00
parent cc95590196
commit c5ea3a7b59
8 changed files with 1429 additions and 0 deletions
@@ -0,0 +1,504 @@
+using System.Diagnostics;
+using System.Reflection;
+using Grpc.Net.Client;
+using Microsoft.Extensions.Configuration;
+using Microsoft.Extensions.DependencyInjection;
+using Microsoft.Extensions.Logging;
+using Microsoft.Extensions.Options;
+using ZB.MOM.WW.Secrets.Abstractions;
+using ZB.MOM.WW.Secrets.Replicator.Grpc.DependencyInjection;
+using ZB.MOM.WW.Secrets.Sqlite;
+
+namespace ZB.MOM.WW.Secrets.Replicator.Grpc.Tests;
+
+/// <summary>
+/// Convergence behaviour of the follower's pull-only sweep, exercised against real SQLite stores on
+/// both sides so the assertions are about actual persisted state rather than mock interactions.
+/// </summary>
+public sealed class GrpcSecretSyncServiceTests : IDisposable
+{
+    private readonly List<string> _dbPaths = [];
+    private readonly SqliteSecretStore _local;
+    private readonly SqliteSecretStore _hubStore;
+    private readonly FakeHubReader _hub;
+    private readonly RecordingLogger _logger = new();
+    private readonly RecordingCacheInvalidator _cache = new();
+
+    public GrpcSecretSyncServiceTests()
+    {
+        _local = CreateStore();
+        _hubStore = CreateStore();
+        _hub = new FakeHubReader(_hubStore);
+    }
+
+    private SqliteSecretStore CreateStore()
+    {
+        string path = Path.Combine(Path.GetTempPath(), $"zb-grpc-sync-{Guid.NewGuid():N}.db");
+        _dbPaths.Add(path);
+        var factory = new SecretsSqliteConnectionFactory(path);
+        new SqliteSecretsStoreMigrator(factory).MigrateAsync(CancellationToken.None).GetAwaiter().GetResult();
+        return new SqliteSecretStore(factory);
+    }
+
+    private GrpcSecretSyncService CreateService(TimeSpan? interval = null) => new(
+        _local,
+        _hub,
+        _cache,
+        Options.Create(new SecretsGrpcHubClientOptions
+        {
+            Endpoint = "https://central:5001",
+            BearerToken = "unused-by-the-fake",
+            SyncInterval = interval ?? TimeSpan.FromHours(1),
+        }),
+        _logger);
+
+    private static StoredSecret Row(string name, byte marker) => new()
+    {
+        Name = new SecretName(name),
+        ContentType = SecretContentType.Text,
+        Ciphertext = [marker],
+        Nonce = [1],
+        Tag = [2],
+        WrappedDek = [3],
+        WrapNonce = [4],
+        WrapTag = [5],
+        KekId = "sha256:test",
+        Revision = 0,
+        CreatedUtc = DateTimeOffset.UtcNow,
+        UpdatedUtc = DateTimeOffset.UtcNow,
+    };
+
+    [Fact]
+    public async Task Sweep_pulls_a_secret_the_node_has_never_seen()
+    {
+        await _hubStore.UpsertAsync(Row("db/password", 0xAA), CancellationToken.None);
+
+        int pulled = await CreateService().SweepAsync(CancellationToken.None);
+
+        Assert.Equal(1, pulled);
+
+        StoredSecret? local = await _local.GetAsync(new SecretName("db/password"), CancellationToken.None);
+        Assert.NotNull(local);
+        Assert.Equal(0xAA, local!.Ciphertext[0]);
+    }
+
+    [Fact]
+    public async Task Sweep_pulls_the_row_byte_identically()
+    {
+        await _hubStore.UpsertAsync(Row("db/password", 0xAA), CancellationToken.None);
+        await _hubStore.UpsertAsync(Row("db/password", 0xBB), CancellationToken.None);
+
+        StoredSecret hubRow = (await _hubStore.GetAsync(new SecretName("db/password"), CancellationToken.None))!;
+
+        await CreateService().SweepAsync(CancellationToken.None);
+
+        StoredSecret localRow = (await _local.GetAsync(new SecretName("db/password"), CancellationToken.None))!;
+
+        Assert.Equal(hubRow.Ciphertext, localRow.Ciphertext);
+        Assert.Equal(hubRow.WrappedDek, localRow.WrappedDek);
+        Assert.Equal(hubRow.KekId, localRow.KekId);
+        Assert.Equal(hubRow.Revision, localRow.Revision);
+        Assert.Equal(hubRow.UpdatedUtc, localRow.UpdatedUtc);
+    }
+
+    [Fact]
+    public async Task Sweep_applies_a_hub_tombstone_over_a_live_local_row()
+    {
+        await _hubStore.UpsertAsync(Row("api/key", 0xBB), CancellationToken.None);
+        await CreateService().SweepAsync(CancellationToken.None);
+
+        await _hubStore.DeleteAsync(new SecretName("api/key"), "admin", CancellationToken.None);
+        int pulled = await CreateService().SweepAsync(CancellationToken.None);
+
+        Assert.Equal(1, pulled);
+
+        StoredSecret localRow = (await _local.GetAsync(new SecretName("api/key"), CancellationToken.None))!;
+        Assert.True(localRow.IsDeleted);
+    }
+
+    [Fact]
+    public async Task Sweep_is_idempotent_so_a_converged_follower_stays_quiet()
+    {
+        await _hubStore.UpsertAsync(Row("api/key", 0xBB), CancellationToken.None);
+
+        GrpcSecretSyncService service = CreateService();
+        await service.SweepAsync(CancellationToken.None);
+
+        int pulled = await service.SweepAsync(CancellationToken.None);
+
+        Assert.Equal(0, pulled);
+    }
+
+    [Fact]
+    public async Task Sweep_skips_the_fetch_when_the_hub_has_nothing_newer()
+    {
+        await CreateService().SweepAsync(CancellationToken.None);
+
+        Assert.Equal(0, _hub.GetManyCallCount);
+    }
+
+    [Fact]
+    public async Task Sweep_never_sends_a_local_only_secret_hub_ward()
+    {
+        // Pull-only: a row this node holds and the hub does not stays put. In the SQL-Server
+        // topology this would be pushed; here central is the sole origin of every update, and the
+        // contract has no write RPC to push it with.
+        await _local.UpsertAsync(Row("only/local", 0x11), CancellationToken.None);
+
+        int pulled = await CreateService().SweepAsync(CancellationToken.None);
+
+        Assert.Equal(0, pulled);
+        Assert.Empty(await _hubStore.GetManifestAsync(CancellationToken.None));
+    }
+
+    [Fact]
+    public void The_hub_reader_seam_exposes_no_write_operation()
+    {
+        // The structural half of pull-only. The proto has no write RPC; this pins that the client's
+        // own seam cannot grow one either, so a future edit has to break this test on purpose.
+        string[] members = typeof(ISecretsHubReader)
+            .GetMethods(BindingFlags.Public | BindingFlags.Instance | BindingFlags.DeclaredOnly)
+            .Select(m => m.Name)
+            .Order(StringComparer.Ordinal)
+            .ToArray();
+
+        Assert.Equal(["GetManifestAsync", "GetManyAsync"], members);
+    }
+
+    [Theory]
+    [InlineData("ComputePushSet")]
+    [InlineData("ReadLocalAsync")]
+    [InlineData("ApplyReplicatedAsync")]
+    public void The_client_package_never_names_a_hub_ward_write_path(string forbidden)
+    {
+        // ComputePushSet and ReadLocalAsync are the two reconciler entry points that assemble rows
+        // to move toward a peer; ApplyReplicatedAsync on a hub-side store is how they would land.
+        // Pull-only is an owner-confirmed invariant, so the guard is a source scan rather than a
+        // behavioural test: a future edit has to delete this test on purpose to reintroduce a push.
+        foreach (string file in ClientSourceFiles())
+        {
+            string source = File.ReadAllText(file);
+
+            Assert.DoesNotContain(
+                forbidden,
+                StripComments(source),
+                StringComparison.Ordinal);
+        }
+    }
+
+    // The client half of the package — deliberately excludes the server-side files, which legitimately
+    // read a local store to serve the hub.
+    private static IEnumerable<string> ClientSourceFiles()
+    {
+        string root = PackageSourceRoot();
+
+        string[] files =
+        [
+            Path.Combine(root, "GrpcSecretSyncService.cs"),
+            Path.Combine(root, "GrpcSecretsHubClient.cs"),
+            Path.Combine(root, "Wire", "WireToDomain.cs"),
+            Path.Combine(root, "DependencyInjection", "SecretsGrpcHubClientExtensions.cs"),
+        ];
+
+        foreach (string file in files)
+        {
+            Assert.True(File.Exists(file), $"expected client source file is missing: {file}");
+        }
+
+        return files;
+    }
+
+    private static string PackageSourceRoot()
+    {
+        DirectoryInfo? dir = new(AppContext.BaseDirectory);
+
+        while (dir is not null && !File.Exists(Path.Combine(dir.FullName, "ZB.MOM.WW.Secrets.slnx")))
+        {
+            dir = dir.Parent;
+        }
+
+        Assert.NotNull(dir);
+
+        return Path.Combine(dir!.FullName, "src", "ZB.MOM.WW.Secrets.Replicator.Grpc");
+    }
+
+    // Comments explain what the code deliberately does NOT do, and naming the forbidden member is
+    // exactly how that is explained — so the scan looks at code only.
+    private static string StripComments(string source)
+    {
+        IEnumerable<string> lines = source
+            .Split('\n')
+            .Select(line => line.TrimStart())
+            .Where(line => !line.StartsWith("//", StringComparison.Ordinal)
+                && !line.StartsWith("///", StringComparison.Ordinal)
+                && !line.StartsWith('*'));
+
+        return string.Join('\n', lines);
+    }
+
+    [Fact]
+    public async Task Sweep_evicts_the_resolver_cache_for_applied_names_only()
+    {
+        await _hubStore.UpsertAsync(Row("db/password", 0xAA), CancellationToken.None);
+        await _local.UpsertAsync(Row("only/local", 0x11), CancellationToken.None);
+
+        await CreateService().SweepAsync(CancellationToken.None);
+
+        Assert.Equal(["db/password"], _cache.Invalidated.Select(n => n.Value));
+    }
+
+    [Fact]
+    public async Task A_hub_outage_on_the_manifest_call_propagates_out_of_the_raw_sweep()
+    {
+        // Establishes that the fake really fails, so the next two tests prove containment rather
+        // than an accidentally-healthy hub.
+        _hub.FailManifestWith = () => new InvalidOperationException("hub unreachable");
+
+        await Assert.ThrowsAsync<InvalidOperationException>(
+            () => CreateService().SweepAsync(CancellationToken.None));
+    }
+
+    [Fact]
+    public async Task A_hub_outage_on_the_manifest_call_never_escapes_the_background_loop()
+    {
+        await _local.UpsertAsync(Row("api/key", 0xBB), CancellationToken.None);
+        StoredSecret before = (await _local.GetAsync(new SecretName("api/key"), CancellationToken.None))!;
+
+        _hub.FailManifestWith = () => new InvalidOperationException("hub unreachable");
+
+        await RunOneStartupSweepAsync(() => _hub.ManifestCallCount > 0);
+
+        StoredSecret after = (await _local.GetAsync(new SecretName("api/key"), CancellationToken.None))!;
+        Assert.Equal(before.Ciphertext, after.Ciphertext);
+        Assert.Equal(before.Revision, after.Revision);
+        Assert.Equal(before.UpdatedUtc, after.UpdatedUtc);
+        Assert.False(after.IsDeleted);
+        Assert.Equal(1, _logger.WarningCount);
+    }
+
+    [Fact]
+    public async Task A_hub_outage_on_the_fetch_call_never_escapes_the_background_loop()
+    {
+        await _hubStore.UpsertAsync(Row("db/password", 0xAA), CancellationToken.None);
+        _hub.FailFetchWith = () => new InvalidOperationException("hub unreachable mid-fetch");
+
+        await RunOneStartupSweepAsync(() => _hub.GetManyCallCount > 0);
+
+        Assert.Null(await _local.GetAsync(new SecretName("db/password"), CancellationToken.None));
+        Assert.Equal(1, _logger.WarningCount);
+    }
+
+    [Fact]
+    public async Task A_quiet_sweep_logs_nothing_at_information()
+    {
+        await RunOneStartupSweepAsync(() => _hub.ManifestCallCount > 0);
+
+        Assert.Equal(0, _logger.InformationCount);
+        Assert.Equal(0, _logger.WarningCount);
+    }
+
+    [Fact]
+    public async Task A_sweep_that_moved_rows_reports_how_many_it_pulled()
+    {
+        await _hubStore.UpsertAsync(Row("db/password", 0xAA), CancellationToken.None);
+
+        await RunOneStartupSweepAsync(() => _hub.GetManyCallCount > 0);
+
+        Assert.Equal(1, _logger.InformationCount);
+    }
+
+    [Fact]
+    public async Task Startup_sync_can_be_switched_off()
+    {
+        await _hubStore.UpsertAsync(Row("db/password", 0xAA), CancellationToken.None);
+
+        var service = new GrpcSecretSyncService(
+            _local,
+            _hub,
+            _cache,
+            Options.Create(new SecretsGrpcHubClientOptions
+            {
+                Endpoint = "https://central:5001",
+                BearerToken = "unused-by-the-fake",
+                SyncInterval = TimeSpan.FromHours(1),
+                SyncOnStartup = false,
+            }),
+            _logger);
+
+        await service.StartAsync(CancellationToken.None);
+        await service.StopAsync(CancellationToken.None);
+
+        Assert.Equal(0, _hub.ManifestCallCount);
+    }
+
+    [Fact]
+    public void Registration_refuses_a_configuration_with_no_bearer_token()
+    {
+        IConfiguration config = new ConfigurationBuilder()
+            .AddInMemoryCollection(new Dictionary<string, string?>
+            {
+                ["Secrets:GrpcHub:Endpoint"] = "https://central:5001",
+            })
+            .Build();
+
+        InvalidOperationException ex = Assert.Throws<InvalidOperationException>(
+            () => new ServiceCollection().AddZbSecretsGrpcHubClient(config));
+
+        Assert.Contains("Secrets:GrpcHub:BearerToken", ex.Message, StringComparison.Ordinal);
+    }
+
+    [Fact]
+    public void Registration_does_not_hijack_an_application_s_own_grpc_channel()
+    {
+        // GrpcChannel is a framework type, so an application that already has one registered for its
+        // own dependency must keep it — and this package must still dial the hub, not that endpoint.
+        IConfiguration config = new ConfigurationBuilder()
+            .AddInMemoryCollection(new Dictionary<string, string?>
+            {
+                ["Secrets:GrpcHub:Endpoint"] = "https://central:5001",
+                ["Secrets:GrpcHub:BearerToken"] = "token",
+            })
+            .Build();
+
+        using GrpcChannel applicationChannel = GrpcChannel.ForAddress("https://somewhere-else:9999");
+
+        ServiceCollection services = new();
+        services.AddSingleton(applicationChannel);
+        services.AddZbSecretsGrpcHubClient(config);
+
+        using ServiceProvider provider = services.BuildServiceProvider();
+
+        Assert.Same(applicationChannel, provider.GetRequiredService<GrpcChannel>());
+        Assert.NotSame(
+            applicationChannel,
+            provider.GetRequiredKeyedService<GrpcChannel>(
+                SecretsGrpcHubClientExtensions.ChannelServiceKey));
+    }
+
+    // Runs the hosted service through one startup sweep and shuts it down, asserting that nothing
+    // escaped ExecuteAsync. The long sync interval means exactly one sweep happens.
+    private async Task RunOneStartupSweepAsync(Func<bool> sweepObserved)
+    {
+        GrpcSecretSyncService service = CreateService();
+
+        await service.StartAsync(CancellationToken.None);
+
+        var deadline = Stopwatch.StartNew();
+
+        while (!sweepObserved() && deadline.Elapsed < TimeSpan.FromSeconds(10))
+        {
+            await Task.Delay(10);
+        }
+
+        await service.StopAsync(CancellationToken.None);
+
+        Assert.True(sweepObserved(), "the startup sweep never ran");
+        Assert.NotNull(service.ExecuteTask);
+        Assert.False(service.ExecuteTask!.IsFaulted);
+    }
+
+    public void Dispose()
+    {
+        Microsoft.Data.Sqlite.SqliteConnection.ClearAllPools();
+
+        foreach (string db in _dbPaths)
+        {
+            foreach (string path in new[] { db, db + "-wal", db + "-shm" })
+            {
+                if (File.Exists(path))
+                {
+                    File.Delete(path);
+                }
+            }
+        }
+    }
+
+    /// <summary>
+    /// An <see cref="ISecretsHubReader"/> backed by a real SQLite store, standing in for the
+    /// central-hosted hub. Backed by a real store rather than a dictionary for the same reason the
+    /// SQL-Server suite is: the behaviour under test is last-writer-wins convergence, and a
+    /// hand-rolled fake would re-implement the very logic the tests exist to check.
+    /// </summary>
+    private sealed class FakeHubReader(SqliteSecretStore store) : ISecretsHubReader
+    {
+        public int ManifestCallCount { get; private set; }
+
+        public int GetManyCallCount { get; private set; }
+
+        public Func<Exception>? FailManifestWith { get; set; }
+
+        public Func<Exception>? FailFetchWith { get; set; }
+
+        public Task<IReadOnlyList<SecretManifestEntry>> GetManifestAsync(CancellationToken ct)
+        {
+            ManifestCallCount++;
+
+            return FailManifestWith is null
+                ? store.GetManifestAsync(ct)
+                : Task.FromException<IReadOnlyList<SecretManifestEntry>>(FailManifestWith());
+        }
+
+        public async Task<IReadOnlyList<StoredSecret>> GetManyAsync(
+            IReadOnlyList<SecretName> names, CancellationToken ct)
+        {
+            GetManyCallCount++;
+
+            if (FailFetchWith is not null)
+            {
+                throw FailFetchWith();
+            }
+
+            var rows = new List<StoredSecret>(names.Count);
+
+            foreach (SecretName name in names)
+            {
+                StoredSecret? row = await store.GetAsync(name, ct);
+
+                if (row is not null)
+                {
+                    rows.Add(row);
+                }
+            }
+
+            return rows;
+        }
+    }
+
+    private sealed class RecordingCacheInvalidator : ISecretCacheInvalidator
+    {
+        public List<SecretName> Invalidated { get; } = [];
+
+        public void Invalidate(SecretName name) => Invalidated.Add(name);
+    }
+
+    private sealed class RecordingLogger : ILogger<GrpcSecretSyncService>
+    {
+        public int InformationCount { get; private set; }
+
+        public int WarningCount { get; private 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)
+        {
+            switch (logLevel)
+            {
+                case LogLevel.Information:
+                    InformationCount++;
+                    break;
+                case LogLevel.Warning:
+                    WarningCount++;
+                    break;
+                default:
+                    break;
+            }
+        }
+    }
+}