merge(secrets): Task 3 client half — pull-only sweep + per-row parse isolation

Claude-Session: https://claude.ai/code/session_014WNM4vjoVksyyBraTXSZE1
This commit is contained in:
Joseph Doherty
2026-08-07 06:32:04 -04:00
10 changed files with 2005 additions and 0 deletions
@@ -0,0 +1,551 @@
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.Replicator.Grpc.V1;
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<GrpcSecretSyncService> _logger = new();
private readonly RecordingLogger<GrpcSecretsHubClient> _clientLogger = 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 async Task A_malformed_row_never_poisons_the_batch_it_arrived_in()
{
// The failure this guards against is not "one row is lost" — it is that the pull set never
// changes, so the identical malformed row re-fails the identical sweep every interval and
// the node never converges on anything, indistinguishable in the log from a hub outage.
SecretRow bad = TestWire.Row("api/key", 0xBB);
bad.ContentType = "Kerberos";
var invoker = new FakeSecretsHubCallInvoker
{
ManifestResponse = new GetManifestResponse
{
Entries =
{
TestWire.Entry("db/password"), TestWire.Entry("api/key"), TestWire.Entry("svc/token"),
},
},
SecretsResponse = new GetSecretsResponse
{
Secrets = { TestWire.Row("db/password", 0xAA), bad, TestWire.Row("svc/token", 0xCC) },
},
};
int pulled = await CreateServiceOverTheWire(invoker).SweepAsync(CancellationToken.None);
Assert.Equal(2, pulled);
Assert.NotNull(await _local.GetAsync(new SecretName("db/password"), CancellationToken.None));
Assert.NotNull(await _local.GetAsync(new SecretName("svc/token"), CancellationToken.None));
Assert.Null(await _local.GetAsync(new SecretName("api/key"), CancellationToken.None));
Assert.Single(_clientLogger.Warnings);
}
[Fact]
public async Task A_malformed_manifest_entry_never_poisons_the_manifest_it_arrived_in()
{
ManifestEntry bad = TestWire.Entry("api/key");
bad.UpdatedUtc = "not-a-timestamp";
var invoker = new FakeSecretsHubCallInvoker
{
ManifestResponse = new GetManifestResponse
{
Entries = { TestWire.Entry("db/password"), bad, TestWire.Entry("svc/token") },
},
SecretsResponse = new GetSecretsResponse
{
Secrets = { TestWire.Row("db/password", 0xAA), TestWire.Row("svc/token", 0xCC) },
},
};
int pulled = await CreateServiceOverTheWire(invoker).SweepAsync(CancellationToken.None);
Assert.Equal(2, pulled);
Assert.Equal(["db/password", "svc/token"], invoker.LastSecretsRequest!.Names);
Assert.Single(_clientLogger.Warnings);
}
// A sweep wired to the REAL hub client over a fake CallInvoker, so the parse boundary the other
// tests bypass (the FakeHubReader hands over domain types already) is actually exercised.
private GrpcSecretSyncService CreateServiceOverTheWire(FakeSecretsHubCallInvoker invoker)
{
IOptions<SecretsGrpcHubClientOptions> options = Options.Create(new SecretsGrpcHubClientOptions
{
Endpoint = "https://central:5001",
BearerToken = "hub-token",
SyncInterval = TimeSpan.FromHours(1),
});
var client = new GrpcSecretsHubClient(
new SecretsHub.SecretsHubClient(invoker), options, _clientLogger);
return new GrpcSecretSyncService(_local, client, _cache, options, _logger);
}
[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);
}
}
@@ -0,0 +1,233 @@
using Microsoft.Extensions.Options;
using ZB.MOM.WW.Secrets.Abstractions;
using ZB.MOM.WW.Secrets.Replicator.Grpc.V1;
namespace ZB.MOM.WW.Secrets.Replicator.Grpc.Tests;
/// <summary>
/// The real hub client, driven over a fake <c>CallInvoker</c>: what it puts on the wire (credential
/// and deadline) and what it does with what comes back (per-item conversion, per-item isolation).
/// </summary>
/// <remarks>
/// Everything here is deliberately exercised through <c>GrpcSecretsHubClient</c> itself rather than
/// through a fake of the <c>ISecretsHubReader</c> seam. A seam fake proves nothing about the header,
/// the deadline, or the conversion — all three live only in this class, and all three are exactly
/// where a follower silently talks to the hub wrong.
/// </remarks>
public sealed class GrpcSecretsHubClientTests
{
private static readonly DateTimeOffset Now = new(2026, 8, 7, 12, 0, 0, TimeSpan.Zero);
private readonly FakeSecretsHubCallInvoker _invoker = new();
private readonly RecordingLogger<GrpcSecretsHubClient> _logger = new();
private readonly SecretsGrpcHubClientOptions _options = new()
{
Endpoint = "https://central:5001",
BearerToken = "hub-token",
CallDeadline = TimeSpan.FromSeconds(17),
};
private GrpcSecretsHubClient CreateClient() => new(
new SecretsHub.SecretsHubClient(_invoker),
Options.Create(_options),
_logger,
new FixedTimeProvider(Now));
[Fact]
public async Task Every_call_presents_the_bearer_credential()
{
_invoker.ManifestResponse = new GetManifestResponse { Entries = { TestWire.Entry("db/password") } };
_invoker.SecretsResponse = new GetSecretsResponse { Secrets = { TestWire.Row("db/password", 0xAA) } };
GrpcSecretsHubClient client = CreateClient();
await client.GetManifestAsync(CancellationToken.None);
await client.GetManyAsync([new SecretName("db/password")], CancellationToken.None);
Assert.Equal(["GetManifest", "GetSecrets"], _invoker.Methods);
Assert.Equal("Bearer hub-token", _invoker.AuthorizationOn(0));
Assert.Equal("Bearer hub-token", _invoker.AuthorizationOn(1));
}
[Fact]
public async Task Every_call_carries_a_deadline_of_now_plus_the_configured_budget()
{
await CreateClient().GetManifestAsync(CancellationToken.None);
Assert.Equal(Now.UtcDateTime.AddSeconds(17), _invoker.Calls[0].Deadline);
}
[Fact]
public async Task The_deadline_is_recomputed_per_call_rather_than_cached()
{
// A CallOptions built once at construction would carry an absolute deadline that expires
// shortly after startup and then fails every sweep forever, which reads as a hub outage.
GrpcSecretsHubClient client = CreateClient();
await client.GetManifestAsync(CancellationToken.None);
await client.GetManifestAsync(CancellationToken.None);
Assert.Equal(_invoker.Calls[0].Deadline, _invoker.Calls[1].Deadline);
Assert.NotSame(_invoker.Calls[0].Headers, _invoker.Calls[1].Headers);
}
[Fact]
public async Task The_fetch_asks_for_exactly_the_names_it_was_given()
{
_invoker.SecretsResponse = new GetSecretsResponse();
await CreateClient().GetManyAsync(
[new SecretName("db/password"), new SecretName("api/key")], CancellationToken.None);
Assert.Equal(["db/password", "api/key"], _invoker.LastSecretsRequest!.Names);
}
[Fact]
public async Task A_wire_row_comes_back_as_a_domain_row()
{
_invoker.SecretsResponse = new GetSecretsResponse { Secrets = { TestWire.Row("db/password", 0xAA) } };
IReadOnlyList<StoredSecret> rows =
await CreateClient().GetManyAsync([new SecretName("db/password")], CancellationToken.None);
StoredSecret row = Assert.Single(rows);
Assert.Equal("db/password", row.Name.Value);
Assert.Equal(SecretContentType.Text, row.ContentType);
Assert.Equal(new byte[] { 0xAA }, row.Ciphertext);
Assert.Equal(TestWire.FixedInstant, row.UpdatedUtc);
}
[Fact]
public async Task A_wire_manifest_entry_comes_back_as_a_domain_entry()
{
_invoker.ManifestResponse = new GetManifestResponse { Entries = { TestWire.Entry("api/key") } };
IReadOnlyList<SecretManifestEntry> entries =
await CreateClient().GetManifestAsync(CancellationToken.None);
SecretManifestEntry entry = Assert.Single(entries);
Assert.Equal("api/key", entry.Name.Value);
Assert.Equal(TestWire.FixedInstant, entry.UpdatedUtc);
}
[Fact]
public async Task A_malformed_row_is_dropped_and_its_well_formed_siblings_survive()
{
// The whole point: one unparseable row must not cost the batch. Eagerly materializing the
// response would fail all three, and the identical pull set would re-fail every interval.
SecretRow bad = TestWire.Row("api/key", 0xBB);
bad.ContentType = "Kerberos";
_invoker.SecretsResponse = new GetSecretsResponse
{
Secrets = { TestWire.Row("db/password", 0xAA), bad, TestWire.Row("svc/token", 0xCC) },
};
IReadOnlyList<StoredSecret> rows = await CreateClient().GetManyAsync(
[new SecretName("db/password"), new SecretName("api/key"), new SecretName("svc/token")],
CancellationToken.None);
Assert.Equal(["db/password", "svc/token"], rows.Select(r => r.Name.Value));
}
[Fact]
public async Task A_dropped_row_costs_exactly_one_warning_that_names_it()
{
SecretRow bad = TestWire.Row("api/key", 0xBB);
bad.ContentType = "Kerberos";
_invoker.SecretsResponse = new GetSecretsResponse
{
Secrets = { TestWire.Row("db/password", 0xAA), bad },
};
await CreateClient().GetManyAsync(
[new SecretName("db/password"), new SecretName("api/key")], CancellationToken.None);
string warning = Assert.Single(_logger.Warnings);
Assert.Contains("api/key", warning, StringComparison.Ordinal);
}
[Fact]
public async Task A_dropped_row_never_logs_the_material_it_was_carrying()
{
// A parse failure is an operations problem, so it is logged — but a replicated row is
// ciphertext plus its wrapped DEK, and none of that may reach a log sink.
SecretRow bad = TestWire.Row("api/key", 0xBB);
bad.UpdatedUtc = "yesterday";
_invoker.SecretsResponse = new GetSecretsResponse { Secrets = { bad } };
await CreateClient().GetManyAsync([new SecretName("api/key")], CancellationToken.None);
string warning = Assert.Single(_logger.Warnings);
foreach (string forbidden in new[] { "sha256:test", "hub-token", "Ciphertext", "WrappedDek" })
{
Assert.DoesNotContain(forbidden, warning, StringComparison.OrdinalIgnoreCase);
}
}
[Fact]
public async Task A_row_whose_own_name_is_unparseable_is_dropped_without_echoing_it()
{
// The name is the one field a warning may legitimately carry — but here the name IS what
// failed to parse, so it is replaced by a marker rather than trusted into the log line.
SecretRow bad = TestWire.Row("db/password", 0xAA);
bad.Name = "../../etc/passwd";
_invoker.SecretsResponse = new GetSecretsResponse { Secrets = { bad } };
IReadOnlyList<StoredSecret> rows =
await CreateClient().GetManyAsync([new SecretName("db/password")], CancellationToken.None);
Assert.Empty(rows);
string warning = Assert.Single(_logger.Warnings);
Assert.DoesNotContain("etc/passwd", warning, StringComparison.Ordinal);
}
[Fact]
public async Task A_malformed_manifest_entry_is_dropped_and_the_rest_of_the_manifest_survives()
{
ManifestEntry bad = TestWire.Entry("api/key");
bad.UpdatedUtc = "not-a-timestamp";
_invoker.ManifestResponse = new GetManifestResponse
{
Entries = { TestWire.Entry("db/password"), bad, TestWire.Entry("svc/token") },
};
IReadOnlyList<SecretManifestEntry> entries =
await CreateClient().GetManifestAsync(CancellationToken.None);
Assert.Equal(["db/password", "svc/token"], entries.Select(e => e.Name.Value));
Assert.Single(_logger.Warnings);
}
[Fact]
public async Task Conversion_preserves_the_order_the_hub_sent()
{
_invoker.SecretsResponse = new GetSecretsResponse
{
Secrets = { TestWire.Row("db/password", 0xAA), TestWire.Row("api/key", 0xBB) },
};
IReadOnlyList<StoredSecret> rows = await CreateClient().GetManyAsync(
[new SecretName("db/password"), new SecretName("api/key")], CancellationToken.None);
Assert.Equal(["db/password", "api/key"], rows.Select(r => r.Name.Value));
}
[Fact]
public async Task A_well_formed_response_logs_nothing()
{
_invoker.ManifestResponse = new GetManifestResponse { Entries = { TestWire.Entry("db/password") } };
await CreateClient().GetManifestAsync(CancellationToken.None);
Assert.Empty(_logger.Warnings);
Assert.Empty(_logger.Information);
}
}
@@ -0,0 +1,132 @@
namespace ZB.MOM.WW.Secrets.Replicator.Grpc.Tests;
/// <summary>
/// Fail-closed validation of the follower's hub-client options. Every message must name the exact
/// configuration key an operator has to fix, and none of them may ever echo the bearer token.
/// </summary>
public sealed class SecretsGrpcHubClientOptionsTests
{
private const string Token = "s3cr3t-hub-token";
private static SecretsGrpcHubClientOptions Valid() => new()
{
Endpoint = "https://central:5001",
BearerToken = Token,
};
[Fact]
public void Defaults_match_the_documented_contract()
{
var options = new SecretsGrpcHubClientOptions();
Assert.Equal(TimeSpan.FromSeconds(30), options.SyncInterval);
Assert.Equal(TimeSpan.FromSeconds(30), options.CallDeadline);
Assert.True(options.SyncOnStartup);
}
[Fact]
public void A_complete_configuration_validates()
{
Valid().Validate();
}
[Theory]
[InlineData("")]
[InlineData(" ")]
public void A_missing_endpoint_names_its_key(string endpoint)
{
SecretsGrpcHubClientOptions options = Valid();
options.Endpoint = endpoint;
InvalidOperationException ex = Assert.Throws<InvalidOperationException>(() => options.Validate());
Assert.Contains("Secrets:GrpcHub:Endpoint", ex.Message, StringComparison.Ordinal);
}
[Theory]
[InlineData("central:5001")]
[InlineData("/relative/path")]
[InlineData("ftp://central:5001")]
[InlineData("unix:///var/run/hub.sock")]
public void An_endpoint_that_is_not_an_absolute_http_uri_names_its_key(string endpoint)
{
SecretsGrpcHubClientOptions options = Valid();
options.Endpoint = endpoint;
InvalidOperationException ex = Assert.Throws<InvalidOperationException>(() => options.Validate());
Assert.Contains("Secrets:GrpcHub:Endpoint", ex.Message, StringComparison.Ordinal);
}
[Theory]
[InlineData("")]
[InlineData(" ")]
public void A_missing_bearer_token_names_its_key(string token)
{
SecretsGrpcHubClientOptions options = Valid();
options.BearerToken = token;
InvalidOperationException ex = Assert.Throws<InvalidOperationException>(() => options.Validate());
Assert.Contains("Secrets:GrpcHub:BearerToken", ex.Message, StringComparison.Ordinal);
}
public static TheoryData<Action<SecretsGrpcHubClientOptions>> InvalidMutations() =>
[
o => o.Endpoint = string.Empty,
o => o.Endpoint = "not a uri",
o => o.BearerToken = string.Empty,
o => o.SyncInterval = TimeSpan.Zero,
o => o.CallDeadline = TimeSpan.FromSeconds(-1),
];
[Theory]
[MemberData(nameof(InvalidMutations))]
public void No_validation_message_ever_echoes_the_bearer_token(
Action<SecretsGrpcHubClientOptions> makeInvalid)
{
// A configuration error surfaces in logs and crash dumps. Naming the key is the whole job;
// quoting the value would leak the hub credential to everyone who can read a startup log.
SecretsGrpcHubClientOptions options = Valid();
makeInvalid(options);
InvalidOperationException ex = Assert.Throws<InvalidOperationException>(() => options.Validate());
Assert.DoesNotContain(Token, ex.Message, StringComparison.Ordinal);
}
[Theory]
[InlineData(0)]
[InlineData(-1)]
public void A_non_positive_sync_interval_names_its_key(int seconds)
{
SecretsGrpcHubClientOptions options = Valid();
options.SyncInterval = TimeSpan.FromSeconds(seconds);
InvalidOperationException ex = Assert.Throws<InvalidOperationException>(() => options.Validate());
Assert.Contains("Secrets:GrpcHub:SyncInterval", ex.Message, StringComparison.Ordinal);
}
[Theory]
[InlineData(0)]
[InlineData(-1)]
public void A_non_positive_call_deadline_names_its_key(int seconds)
{
SecretsGrpcHubClientOptions options = Valid();
options.CallDeadline = TimeSpan.FromSeconds(seconds);
InvalidOperationException ex = Assert.Throws<InvalidOperationException>(() => options.Validate());
Assert.Contains("Secrets:GrpcHub:CallDeadline", ex.Message, StringComparison.Ordinal);
}
[Fact]
public void Messages_name_the_section_the_caller_actually_bound()
{
// The registration extension takes a section path, so a deployment that binds somewhere else
// must be told about ITS key, not the default one.
SecretsGrpcHubClientOptions options = Valid();
options.BearerToken = string.Empty;
InvalidOperationException ex = Assert.Throws<InvalidOperationException>(
() => options.Validate("Central:Secrets:Hub"));
Assert.Contains("Central:Secrets:Hub:BearerToken", ex.Message, StringComparison.Ordinal);
}
}
@@ -0,0 +1,186 @@
using Google.Protobuf;
using Grpc.Core;
using Microsoft.Extensions.Logging;
using ZB.MOM.WW.Secrets.Abstractions;
using ZB.MOM.WW.Secrets.Replicator.Grpc.V1;
namespace ZB.MOM.WW.Secrets.Replicator.Grpc.Tests;
/// <summary>
/// A <see cref="CallInvoker"/> that answers the two hub RPCs from canned responses and records what
/// the client sent.
/// </summary>
/// <remarks>
/// This is what lets the tests drive the <b>real</b> <c>GrpcSecretsHubClient</c> — the generated
/// client accepts a <see cref="CallInvoker"/>, so substituting one exercises the header, the
/// deadline and the wire-to-domain conversion without a live server. A fake of the
/// <c>ISecretsHubReader</c> seam would bypass every one of those.
/// </remarks>
internal sealed class FakeSecretsHubCallInvoker : CallInvoker
{
/// <summary>Response returned for <c>GetManifest</c>.</summary>
public GetManifestResponse ManifestResponse { get; set; } = new();
/// <summary>Response returned for <c>GetSecrets</c>.</summary>
public GetSecretsResponse SecretsResponse { get; set; } = new();
/// <summary>The options every call was invoked with, in call order.</summary>
public List<CallOptions> Calls { get; } = [];
/// <summary>The RPC names invoked, in call order.</summary>
public List<string> Methods { get; } = [];
/// <summary>The most recent <c>GetSecrets</c> request, so tests can assert what was asked for.</summary>
public GetSecretsRequest? LastSecretsRequest { get; private set; }
/// <summary>Returns the <c>Authorization</c> header value seen on the call at the given index.</summary>
/// <param name="index">Zero-based call index.</param>
/// <returns>The header value, or <see langword="null"/> when the header was not sent.</returns>
public string? AuthorizationOn(int index) =>
Calls[index].Headers?.GetValue("authorization");
/// <inheritdoc />
public override AsyncUnaryCall<TResponse> AsyncUnaryCall<TRequest, TResponse>(
Method<TRequest, TResponse> method, string? host, CallOptions options, TRequest request)
{
ArgumentNullException.ThrowIfNull(method);
Calls.Add(options);
Methods.Add(method.Name);
if (request is GetSecretsRequest secretsRequest)
{
LastSecretsRequest = secretsRequest;
}
object response = method.Name switch
{
"GetManifest" => ManifestResponse,
"GetSecrets" => SecretsResponse,
_ => throw new InvalidOperationException($"unexpected RPC '{method.Name}'."),
};
return new AsyncUnaryCall<TResponse>(
Task.FromResult((TResponse)response),
Task.FromResult(new Metadata()),
() => Status.DefaultSuccess,
() => new Metadata(),
() => { });
}
/// <inheritdoc />
public override TResponse BlockingUnaryCall<TRequest, TResponse>(
Method<TRequest, TResponse> method, string? host, CallOptions options, TRequest request) =>
throw new NotSupportedException("The hub client only makes async unary calls.");
/// <inheritdoc />
public override AsyncServerStreamingCall<TResponse> AsyncServerStreamingCall<TRequest, TResponse>(
Method<TRequest, TResponse> method, string? host, CallOptions options, TRequest request) =>
throw new NotSupportedException("The hub contract has no streaming RPC.");
/// <inheritdoc />
public override AsyncClientStreamingCall<TRequest, TResponse> AsyncClientStreamingCall<TRequest, TResponse>(
Method<TRequest, TResponse> method, string? host, CallOptions options) =>
throw new NotSupportedException("The hub contract has no streaming RPC.");
/// <inheritdoc />
public override AsyncDuplexStreamingCall<TRequest, TResponse> AsyncDuplexStreamingCall<TRequest, TResponse>(
Method<TRequest, TResponse> method, string? host, CallOptions options) =>
throw new NotSupportedException("The hub contract has no streaming RPC.");
}
/// <summary>A clock frozen at a caller-chosen instant, so a deadline assertion can be exact.</summary>
/// <param name="now">The instant every read returns.</param>
internal sealed class FixedTimeProvider(DateTimeOffset now) : TimeProvider
{
/// <inheritdoc />
public override DateTimeOffset GetUtcNow() => now;
}
/// <summary>Captures level and formatted message for every log entry.</summary>
/// <typeparam name="TCategory">The logger category.</typeparam>
internal sealed class RecordingLogger<TCategory> : ILogger<TCategory>
{
/// <summary>Formatted messages logged at <see cref="LogLevel.Information"/>.</summary>
public List<string> Information { get; } = [];
/// <summary>Formatted messages logged at <see cref="LogLevel.Warning"/>.</summary>
public List<string> Warnings { get; } = [];
/// <summary>Count of information entries.</summary>
public int InformationCount => Information.Count;
/// <summary>Count of warning entries.</summary>
public int WarningCount => Warnings.Count;
/// <inheritdoc />
public IDisposable? BeginScope<TState>(TState state) where TState : notnull => null;
/// <inheritdoc />
public bool IsEnabled(LogLevel logLevel) => true;
/// <inheritdoc />
public void Log<TState>(
LogLevel logLevel,
EventId eventId,
TState state,
Exception? exception,
Func<TState, Exception?, string> formatter)
{
ArgumentNullException.ThrowIfNull(formatter);
string message = formatter(state, exception);
switch (logLevel)
{
case LogLevel.Information:
Information.Add(message);
break;
case LogLevel.Warning:
Warnings.Add(message);
break;
default:
break;
}
}
}
/// <summary>Builders for well-formed wire messages that individual tests then damage.</summary>
internal static class TestWire
{
/// <summary>A well-formed wire row.</summary>
/// <param name="name">The secret name.</param>
/// <param name="marker">A byte written into the ciphertext so the row is identifiable.</param>
/// <returns>The wire row.</returns>
public static SecretRow Row(string name, byte marker) => new()
{
Name = name,
ContentType = nameof(SecretContentType.Text),
Ciphertext = ByteString.CopyFrom(marker),
Nonce = ByteString.CopyFrom((byte)1),
Tag = ByteString.CopyFrom((byte)2),
WrappedDek = ByteString.CopyFrom((byte)3),
WrapNonce = ByteString.CopyFrom((byte)4),
WrapTag = ByteString.CopyFrom((byte)5),
KekId = "sha256:test",
Revision = 0,
IsDeleted = false,
CreatedUtc = FixedInstant.ToString("O"),
UpdatedUtc = FixedInstant.ToString("O"),
};
/// <summary>A well-formed manifest entry matching <see cref="Row"/>.</summary>
/// <param name="name">The secret name.</param>
/// <returns>The wire manifest entry.</returns>
public static ManifestEntry Entry(string name) => new()
{
Name = name,
Revision = 0,
UpdatedUtc = FixedInstant.ToString("O"),
IsDeleted = false,
};
/// <summary>The timestamp every builder stamps, so rows differ only where a test makes them differ.</summary>
public static readonly DateTimeOffset FixedInstant =
new(2026, 8, 7, 1, 2, 3, TimeSpan.Zero);
}
@@ -0,0 +1,210 @@
using System.Reflection;
using Google.Protobuf;
using ZB.MOM.WW.Secrets.Abstractions;
using ZB.MOM.WW.Secrets.Replicator.Grpc.V1;
using ZB.MOM.WW.Secrets.Replicator.Grpc.Wire;
namespace ZB.MOM.WW.Secrets.Replicator.Grpc.Tests;
/// <summary>
/// The proto -> domain boundary: a follower rebuilding rows a hub sent it. These are the checks
/// protobuf itself cannot make — a name that normalizes and validates, a content type that names a
/// real member, and timestamps whose exact representation survives, because last-writer-wins
/// compares them.
/// </summary>
public sealed class WireToDomainTests
{
private static SecretRow FullRow() => new()
{
Name = "db/password",
Description = "primary db",
ContentType = nameof(SecretContentType.ConnectionString),
Ciphertext = ByteString.CopyFrom(0xAA, 0xBB),
Nonce = ByteString.CopyFrom(1, 2, 3),
Tag = ByteString.CopyFrom(4, 5),
WrappedDek = ByteString.CopyFrom(6),
WrapNonce = ByteString.CopyFrom(7),
WrapTag = ByteString.CopyFrom(8),
KekId = "sha256:abc",
Revision = 7,
IsDeleted = false,
CreatedUtc = new DateTimeOffset(2026, 8, 7, 1, 2, 3, TimeSpan.Zero).ToString("O"),
UpdatedUtc = new DateTimeOffset(2026, 8, 7, 4, 5, 6, TimeSpan.Zero).ToString("O"),
CreatedBy = "alice",
UpdatedBy = "bob",
};
[Fact]
public void Row_maps_every_field_onto_the_domain_type()
{
StoredSecret row = WireToDomain.ToRow(FullRow());
Assert.Equal("db/password", row.Name.Value);
Assert.Equal("primary db", row.Description);
Assert.Equal(SecretContentType.ConnectionString, row.ContentType);
Assert.Equal(new byte[] { 0xAA, 0xBB }, row.Ciphertext);
Assert.Equal(new byte[] { 1, 2, 3 }, row.Nonce);
Assert.Equal(new byte[] { 4, 5 }, row.Tag);
Assert.Equal(new byte[] { 6 }, row.WrappedDek);
Assert.Equal(new byte[] { 7 }, row.WrapNonce);
Assert.Equal(new byte[] { 8 }, row.WrapTag);
Assert.Equal("sha256:abc", row.KekId);
Assert.Equal(7, row.Revision);
Assert.False(row.IsDeleted);
Assert.Null(row.DeletedUtc);
Assert.Equal("alice", row.CreatedBy);
Assert.Equal("bob", row.UpdatedBy);
}
[Fact]
public void Row_rejects_an_unrecognized_content_type()
{
// The reason the wire carries a NAME rather than a proto enum: an unknown value is rejected
// at the boundary instead of silently decoding to the zero value and retyping the secret.
SecretRow wire = FullRow();
wire.ContentType = "Kerberos";
ArgumentException ex = Assert.Throws<ArgumentException>(() => WireToDomain.ToRow(wire));
Assert.Contains("Kerberos", ex.Message, StringComparison.Ordinal);
}
[Fact]
public void Row_rejects_a_numeric_content_type_that_names_no_member()
{
SecretRow wire = FullRow();
wire.ContentType = "9999";
Assert.Throws<ArgumentException>(() => WireToDomain.ToRow(wire));
}
[Fact]
public void Row_rejects_a_name_the_domain_would_not_accept()
{
SecretRow wire = FullRow();
wire.ContentType = nameof(SecretContentType.Text);
wire.Name = "../../etc/passwd";
Assert.Throws<ArgumentException>(() => WireToDomain.ToRow(wire));
}
[Fact]
public void Row_distinguishes_an_absent_description_from_an_empty_one()
{
SecretRow absent = FullRow();
absent.ClearDescription();
Assert.Null(WireToDomain.ToRow(absent).Description);
SecretRow empty = FullRow();
empty.Description = string.Empty;
Assert.Equal(string.Empty, WireToDomain.ToRow(empty).Description);
}
[Fact]
public void Row_distinguishes_absent_from_empty_for_every_nullable_field()
{
SecretRow absent = FullRow();
absent.ClearCreatedBy();
absent.ClearUpdatedBy();
StoredSecret row = WireToDomain.ToRow(absent);
Assert.Null(row.CreatedBy);
Assert.Null(row.UpdatedBy);
}
[Fact]
public void Row_round_trips_a_non_utc_offset_exactly()
{
// The whole reason the contract uses "O" strings and not google.protobuf.Timestamp: an
// offset-bearing DateTimeOffset must come back byte-identical, because LWW compares it.
var updated = new DateTimeOffset(2026, 8, 7, 4, 5, 6, 789, TimeSpan.FromHours(-5));
SecretRow wire = FullRow();
wire.UpdatedUtc = updated.ToString("O");
StoredSecret row = WireToDomain.ToRow(wire);
Assert.Equal(updated, row.UpdatedUtc);
Assert.Equal(updated.Offset, row.UpdatedUtc.Offset);
Assert.Equal(updated.ToString("O"), row.UpdatedUtc.ToString("O"));
}
[Fact]
public void Row_maps_a_tombstone_with_its_deletion_timestamp()
{
var deleted = new DateTimeOffset(2026, 8, 7, 9, 9, 9, TimeSpan.Zero);
SecretRow wire = FullRow();
wire.IsDeleted = true;
wire.DeletedUtc = deleted.ToString("O");
StoredSecret row = WireToDomain.ToRow(wire);
Assert.True(row.IsDeleted);
Assert.Equal(deleted, row.DeletedUtc);
}
[Fact]
public void Row_leaves_the_deletion_timestamp_null_when_the_field_is_absent()
{
SecretRow wire = FullRow();
wire.ClearDeletedUtc();
Assert.Null(WireToDomain.ToRow(wire).DeletedUtc);
}
[Fact]
public void Row_rejects_a_malformed_timestamp()
{
SecretRow wire = FullRow();
wire.UpdatedUtc = "yesterday";
Assert.Throws<FormatException>(() => WireToDomain.ToRow(wire));
}
[Fact]
public void Manifest_entry_maps_and_validates_its_name()
{
var updated = new DateTimeOffset(2026, 8, 7, 4, 5, 6, TimeSpan.FromHours(2));
SecretManifestEntry entry = WireToDomain.ToEntry(new ManifestEntry
{
Name = "api/key",
Revision = 3,
UpdatedUtc = updated.ToString("O"),
IsDeleted = true,
});
Assert.Equal("api/key", entry.Name.Value);
Assert.Equal(3, entry.Revision);
Assert.Equal(updated, entry.UpdatedUtc);
Assert.True(entry.IsDeleted);
}
[Fact]
public void Manifest_entry_rejects_a_name_the_domain_would_not_accept()
{
Assert.Throws<ArgumentException>(() => WireToDomain.ToEntry(new ManifestEntry
{
Name = "/rooted",
Revision = 1,
UpdatedUtc = DateTimeOffset.UtcNow.ToString("O"),
}));
}
[Fact]
public void There_is_no_batch_entry_point_to_convert_a_whole_response_with()
{
// A batch method here would fail a whole response on one bad element, and since the pull set
// is derived from the hub's manifest the identical bad element would re-fail the identical
// sweep forever. Isolation belongs to GrpcSecretsHubClient, which converts item by item, so
// this pins that the strict parser offers nothing that could poison a batch again.
string[] members = typeof(WireToDomain)
.GetMethods(BindingFlags.Public | BindingFlags.Static | BindingFlags.DeclaredOnly)
.Select(m => m.Name)
.Order(StringComparer.Ordinal)
.ToArray();
Assert.Equal(["ToEntry", "ToRow"], members);
}
}