feat(site): DeployedSecretsMirror — per-node apply + decrypt-verify of deployed secrets
This commit is contained in:
Joseph Doherty
@@ -0,0 +1,446 @@
|
||||
using System.Collections.Concurrent;
|
||||
using Microsoft.Extensions.Logging;
|
||||
using Microsoft.Extensions.Logging.Abstractions;
|
||||
using Microsoft.Extensions.Options;
|
||||
using ZB.MOM.WW.ScadaBridge.Commons.Messages.Artifacts;
|
||||
using ZB.MOM.WW.ScadaBridge.SiteRuntime.Persistence;
|
||||
using ZB.MOM.WW.ScadaBridge.SiteRuntime.Secrets;
|
||||
using ZB.MOM.WW.ScadaBridge.TestSupport;
|
||||
using ZB.MOM.WW.Secrets.Abstractions;
|
||||
|
||||
namespace ZB.MOM.WW.ScadaBridge.SiteRuntime.Tests.Secrets;
|
||||
|
||||
/// <summary>
|
||||
/// Task 9, site-secrets artifact deploy: <see cref="DeployedSecretsMirror"/> — per-node apply
|
||||
/// + decrypt-verify of <c>deployed_secrets</c> into this node's own secret store.
|
||||
/// </summary>
|
||||
public class DeployedSecretsMirrorTests : IDisposable
|
||||
{
|
||||
private readonly TestLocalDb _localDb;
|
||||
|
||||
public DeployedSecretsMirrorTests()
|
||||
{
|
||||
_localDb = TestLocalDb.CreateTemp("deployed-secrets-mirror");
|
||||
}
|
||||
|
||||
public void Dispose()
|
||||
{
|
||||
var path = _localDb.Path;
|
||||
_localDb.Dispose();
|
||||
TestLocalDb.DeleteFiles(path);
|
||||
}
|
||||
|
||||
// ── Apply / remove / out-of-band ──
|
||||
|
||||
[Fact]
|
||||
public async Task Reconcile_Applies_Verifies_AndOnlyDeletesWhatItMirrored()
|
||||
{
|
||||
var (mirror, storage, store, _) = await CreateAsync();
|
||||
store.Seed("out-of-band");
|
||||
|
||||
await storage.UpsertDeployedSecretsAsync([Art("a", 1), Art("out-of-band", 1)], "d1");
|
||||
var r1 = await mirror.ReconcileAsync(CancellationToken.None);
|
||||
|
||||
Assert.Equal(new[] { "a", "out-of-band" }, r1.Applied.OrderBy(x => x, StringComparer.Ordinal));
|
||||
Assert.Empty(r1.Failed);
|
||||
|
||||
await storage.UpsertDeployedSecretsAsync([], "d2");
|
||||
// Simulate: OUT-OF-BAND was never actually mirrored by this node (an operator secret
|
||||
// that happens to share the name) by clearing its ledger row directly.
|
||||
await storage.RemoveMirroredSecretAsync("out-of-band");
|
||||
|
||||
var r2 = await mirror.ReconcileAsync(CancellationToken.None);
|
||||
|
||||
Assert.Equal(["a"], r2.Removed);
|
||||
Assert.True(store.Contains("out-of-band"));
|
||||
Assert.False(store.Contains("a"));
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task Reconcile_ReAppliesOnlyWhenRevisionChanges()
|
||||
{
|
||||
var (mirror, storage, store, _) = await CreateAsync();
|
||||
|
||||
await storage.UpsertDeployedSecretsAsync([Art("a", 1)], "d1");
|
||||
var r1 = await mirror.ReconcileAsync(CancellationToken.None);
|
||||
Assert.Equal(["a"], r1.Applied);
|
||||
|
||||
var r2 = await mirror.ReconcileAsync(CancellationToken.None);
|
||||
Assert.Empty(r2.Applied);
|
||||
Assert.False(r2.Changed);
|
||||
|
||||
await storage.UpsertDeployedSecretsAsync([Art("a", 2)], "d2");
|
||||
var r3 = await mirror.ReconcileAsync(CancellationToken.None);
|
||||
Assert.Equal(["a"], r3.Applied);
|
||||
Assert.Equal(2, store.LastApplied["a"].Revision);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task Reconcile_AppliesThroughApplyReplicatedAsync_MappedFromSecretArtifact()
|
||||
{
|
||||
var (mirror, storage, store, _) = await CreateAsync();
|
||||
await storage.UpsertDeployedSecretsAsync([Art("a", 1, kek: "kek-central")], "d1");
|
||||
|
||||
await mirror.ReconcileAsync(CancellationToken.None);
|
||||
|
||||
var applied = store.LastApplied["a"];
|
||||
Assert.Equal("kek-central", applied.KekId);
|
||||
Assert.Equal(1, applied.Revision);
|
||||
Assert.Equal(SecretContentType.Text, applied.ContentType);
|
||||
}
|
||||
|
||||
[Theory]
|
||||
[InlineData(SecretContentType.Text)]
|
||||
[InlineData(SecretContentType.ConnectionString)]
|
||||
[InlineData(SecretContentType.Json)]
|
||||
[InlineData(SecretContentType.BinaryBase64)]
|
||||
public async Task Reconcile_RoundTripsEveryContentType(SecretContentType contentType)
|
||||
{
|
||||
var (mirror, storage, store, _) = await CreateAsync();
|
||||
await storage.UpsertDeployedSecretsAsync(
|
||||
[Art("a", 1, contentType: contentType.ToString())], "d1");
|
||||
|
||||
await mirror.ReconcileAsync(CancellationToken.None);
|
||||
|
||||
Assert.Equal(contentType, store.LastApplied["a"].ContentType);
|
||||
}
|
||||
|
||||
// ── Decrypt verification ──
|
||||
|
||||
[Fact]
|
||||
public async Task Reconcile_DecryptFailure_IsReportedWithKekId()
|
||||
{
|
||||
var (mirror, storage, _, resolver) = await CreateAsync();
|
||||
resolver.FailFor("a");
|
||||
|
||||
await storage.UpsertDeployedSecretsAsync([Art("a", 1, kek: "kek-central")], "d1");
|
||||
var r = await mirror.ReconcileAsync(CancellationToken.None);
|
||||
|
||||
Assert.Contains("kek-central", Assert.Single(r.Failed));
|
||||
Assert.Contains("'a'", Assert.Single(r.Failed));
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task Reconcile_DecryptFailure_NullResolve_IsAlsoReported()
|
||||
{
|
||||
var (mirror, storage, _, resolver) = await CreateAsync();
|
||||
resolver.ReturnNullFor("a");
|
||||
|
||||
await storage.UpsertDeployedSecretsAsync([Art("a", 1)], "d1");
|
||||
var r = await mirror.ReconcileAsync(CancellationToken.None);
|
||||
|
||||
Assert.Single(r.Failed);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task Reconcile_DecryptFailure_ExceptionFromResolver_IsAlsoReported()
|
||||
{
|
||||
var (mirror, storage, _, resolver) = await CreateAsync();
|
||||
resolver.ThrowFor("a");
|
||||
|
||||
await storage.UpsertDeployedSecretsAsync([Art("a", 1)], "d1");
|
||||
var r = await mirror.ReconcileAsync(CancellationToken.None);
|
||||
|
||||
Assert.Single(r.Failed);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task Reconcile_VerifiesEveryPass_EvenWhenAlreadyApplied()
|
||||
{
|
||||
var (mirror, storage, _, resolver) = await CreateAsync();
|
||||
await storage.UpsertDeployedSecretsAsync([Art("a", 1)], "d1");
|
||||
|
||||
var r1 = await mirror.ReconcileAsync(CancellationToken.None);
|
||||
Assert.Empty(r1.Failed);
|
||||
|
||||
// Master key "changes" (or a KEK rotation invalidates the local unwrap) between
|
||||
// passes — even though nothing about deployed_secrets changed, the next reconcile
|
||||
// must still catch the failure because it re-verifies unconditionally.
|
||||
resolver.FailFor("a");
|
||||
var r2 = await mirror.ReconcileAsync(CancellationToken.None);
|
||||
Assert.Single(r2.Failed);
|
||||
Assert.Empty(r2.Applied); // unchanged revision — not re-applied, only re-verified
|
||||
}
|
||||
|
||||
// ── Error-log de-spam (ruling 1) ──
|
||||
|
||||
[Fact]
|
||||
public async Task Reconcile_DecryptFailure_LogsErrorOnlyFirstTime_ThenDebug_ThenErrorAgainAfterASuccess()
|
||||
{
|
||||
var logger = new CapturingLogger<DeployedSecretsMirror>();
|
||||
var (mirror, storage, _, resolver) = await CreateAsync(logger);
|
||||
resolver.FailFor("a");
|
||||
await storage.UpsertDeployedSecretsAsync([Art("a", 1, kek: "kek-1")], "d1");
|
||||
|
||||
await mirror.ReconcileAsync(CancellationToken.None); // 1st failure: Error
|
||||
await mirror.ReconcileAsync(CancellationToken.None); // 2nd identical failure: Debug
|
||||
await mirror.ReconcileAsync(CancellationToken.None); // 3rd identical failure: Debug
|
||||
|
||||
var failureEntries = logger.Entries
|
||||
.Where(e => e.Message.Contains("did not decrypt"))
|
||||
.ToList();
|
||||
Assert.Equal(3, failureEntries.Count);
|
||||
Assert.Equal(LogLevel.Error, failureEntries[0].Level);
|
||||
Assert.Equal(LogLevel.Debug, failureEntries[1].Level);
|
||||
Assert.Equal(LogLevel.Debug, failureEntries[2].Level);
|
||||
|
||||
// Now the secret resolves (e.g. the key issue is fixed).
|
||||
resolver.ClearFailures();
|
||||
var recovered = await mirror.ReconcileAsync(CancellationToken.None);
|
||||
Assert.Empty(recovered.Failed);
|
||||
|
||||
// Fails again — must log at Error again, not Debug, since it succeeded in between.
|
||||
resolver.FailFor("a");
|
||||
await mirror.ReconcileAsync(CancellationToken.None);
|
||||
|
||||
var failureEntriesAfterRecovery = logger.Entries
|
||||
.Where(e => e.Message.Contains("did not decrypt"))
|
||||
.ToList();
|
||||
Assert.Equal(4, failureEntriesAfterRecovery.Count);
|
||||
Assert.Equal(LogLevel.Error, failureEntriesAfterRecovery[3].Level);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task Reconcile_DecryptFailure_NewRevision_LogsErrorAgain_EvenWithoutASuccessInBetween()
|
||||
{
|
||||
var logger = new CapturingLogger<DeployedSecretsMirror>();
|
||||
var (mirror, storage, _, resolver) = await CreateAsync(logger);
|
||||
resolver.FailFor("a");
|
||||
|
||||
await storage.UpsertDeployedSecretsAsync([Art("a", 1, kek: "kek-1")], "d1");
|
||||
await mirror.ReconcileAsync(CancellationToken.None); // Error (rev 1)
|
||||
await mirror.ReconcileAsync(CancellationToken.None); // Debug (rev 1, repeat)
|
||||
|
||||
// Central ships a new revision of the same (still-failing) secret.
|
||||
await storage.UpsertDeployedSecretsAsync([Art("a", 2, kek: "kek-1")], "d2");
|
||||
await mirror.ReconcileAsync(CancellationToken.None); // Error again (rev 2 is new)
|
||||
|
||||
var levels = logger.Entries
|
||||
.Where(e => e.Message.Contains("did not decrypt"))
|
||||
.Select(e => e.Level)
|
||||
.ToList();
|
||||
Assert.Equal([LogLevel.Error, LogLevel.Debug, LogLevel.Error], levels);
|
||||
}
|
||||
|
||||
// ── Hosted loop ──
|
||||
|
||||
[Fact]
|
||||
public async Task ExecuteAsync_RunsOnceAtStart_AndAgainEveryInterval()
|
||||
{
|
||||
var (mirror, storage, store, resolver) = await CreateAsync(
|
||||
interval: TimeSpan.FromMilliseconds(30));
|
||||
await storage.UpsertDeployedSecretsAsync([Art("a", 1)], "d1");
|
||||
|
||||
await mirror.StartAsync(CancellationToken.None);
|
||||
try
|
||||
{
|
||||
await WaitUntilAsync(() => store.Contains("a"));
|
||||
Assert.True(store.Contains("a")); // ran once immediately at start
|
||||
|
||||
// Prove the loop keeps ticking (not just the immediate start-of-day pass) by
|
||||
// observing further resolver invocations over time.
|
||||
var before = resolver.CallCount;
|
||||
await WaitUntilAsync(() => resolver.CallCount > before + 1, timeoutMs: 2000);
|
||||
Assert.True(resolver.CallCount > before + 1);
|
||||
}
|
||||
finally
|
||||
{
|
||||
await mirror.StopAsync(CancellationToken.None);
|
||||
}
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task ExecuteAsync_ReconcileExceptionIsLoggedAndLoopContinues()
|
||||
{
|
||||
var logger = new CapturingLogger<DeployedSecretsMirror>();
|
||||
var (mirror, storage, store, _) = await CreateAsync(
|
||||
logger, interval: TimeSpan.FromMilliseconds(30));
|
||||
await storage.UpsertDeployedSecretsAsync([Art("a", 1)], "d1");
|
||||
store.ThrowOnApply = true;
|
||||
|
||||
await mirror.StartAsync(CancellationToken.None);
|
||||
try
|
||||
{
|
||||
await WaitUntilAsync(() => logger.Entries.Any(
|
||||
e => e.Level == LogLevel.Warning && e.Message.Contains("failed")));
|
||||
|
||||
// The loop must still be alive after the failure — force success and confirm a
|
||||
// later pass applies the secret.
|
||||
store.ThrowOnApply = false;
|
||||
await WaitUntilAsync(() => store.Contains("a"), timeoutMs: 3000);
|
||||
Assert.True(store.Contains("a"));
|
||||
}
|
||||
finally
|
||||
{
|
||||
await mirror.StopAsync(CancellationToken.None);
|
||||
}
|
||||
}
|
||||
|
||||
// ── Harness ──
|
||||
|
||||
private async Task<(DeployedSecretsMirror Mirror, SiteStorageService Storage, FakeSecretStore Store, FakeSecretResolver Resolver)> CreateAsync(
|
||||
CapturingLogger<DeployedSecretsMirror>? logger = null,
|
||||
TimeSpan? interval = null)
|
||||
{
|
||||
var storage = new SiteStorageService(_localDb.Db, NullLogger<SiteStorageService>.Instance);
|
||||
await storage.InitializeAsync();
|
||||
|
||||
var store = new FakeSecretStore();
|
||||
var resolver = new FakeSecretResolver();
|
||||
var options = Options.Create(new SiteRuntimeOptions
|
||||
{
|
||||
DeployedSecretsMirrorInterval = interval ?? TimeSpan.FromSeconds(15),
|
||||
});
|
||||
|
||||
var mirror = new DeployedSecretsMirror(
|
||||
storage,
|
||||
store,
|
||||
resolver,
|
||||
options,
|
||||
(ILogger<DeployedSecretsMirror>?)logger ?? NullLogger<DeployedSecretsMirror>.Instance,
|
||||
TimeProvider.System);
|
||||
|
||||
return (mirror, storage, store, resolver);
|
||||
}
|
||||
|
||||
private static SecretArtifact Art(
|
||||
string name, long revision, string kek = "kek-1", string contentType = "Text") => new(
|
||||
Name: name,
|
||||
ContentType: contentType,
|
||||
Ciphertext: [1, 2, 3, 4, 5],
|
||||
Nonce: [10, 11, 12],
|
||||
Tag: [20, 21, 22, 23],
|
||||
WrappedDek: [30, 31, 32, 33, 34, 35],
|
||||
WrapNonce: [40, 41, 42],
|
||||
WrapTag: [50, 51, 52, 53],
|
||||
KekId: kek,
|
||||
Revision: revision,
|
||||
CreatedUtc: new DateTimeOffset(2026, 9, 28, 0, 0, 0, TimeSpan.Zero),
|
||||
UpdatedUtc: new DateTimeOffset(2026, 9, 28, 1, 0, 0, TimeSpan.Zero),
|
||||
Description: "test secret");
|
||||
|
||||
private static async Task WaitUntilAsync(Func<bool> condition, int timeoutMs = 5000)
|
||||
{
|
||||
var elapsed = 0;
|
||||
const int step = 20;
|
||||
while (!condition() && elapsed < timeoutMs)
|
||||
{
|
||||
await Task.Delay(step);
|
||||
elapsed += step;
|
||||
}
|
||||
Assert.True(condition(), "condition not met within timeout");
|
||||
}
|
||||
|
||||
/// <summary>Thread-safe capturing logger so tests can assert on emitted log levels/messages, mirroring SiteReconciliationActorTests.</summary>
|
||||
private sealed class CapturingLogger<T> : ILogger<T>
|
||||
{
|
||||
public ConcurrentQueue<(LogLevel Level, string Message)> Entries { get; } = new();
|
||||
|
||||
IDisposable? ILogger.BeginScope<TState>(TState state) => null;
|
||||
public bool IsEnabled(LogLevel logLevel) => true;
|
||||
|
||||
public void Log<TState>(
|
||||
LogLevel logLevel, EventId eventId, TState state, Exception? exception,
|
||||
Func<TState, Exception?, string> formatter)
|
||||
=> Entries.Enqueue((logLevel, formatter(state, exception)));
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// In-memory <see cref="ISecretStore"/> recording <see cref="ApplyReplicatedAsync"/> /
|
||||
/// <see cref="DeleteAsync"/> calls, for asserting what the mirror applied/removed without
|
||||
/// a real ZB.MOM.WW.Secrets store.
|
||||
/// </summary>
|
||||
private sealed class FakeSecretStore : ISecretStore
|
||||
{
|
||||
private readonly Dictionary<string, StoredSecret> _rows = new(StringComparer.Ordinal);
|
||||
|
||||
public IReadOnlyDictionary<string, StoredSecret> LastApplied => _rows;
|
||||
|
||||
/// <summary>Set to make the next <see cref="ApplyReplicatedAsync"/> call throw, simulating a store fault.</summary>
|
||||
public bool ThrowOnApply { get; set; }
|
||||
|
||||
/// <summary>Seeds a row directly (bypassing ApplyReplicatedAsync) — an out-of-band secret this node never mirrored itself.</summary>
|
||||
public void Seed(string name) => _rows[name] = new StoredSecret
|
||||
{
|
||||
Name = new SecretName(name),
|
||||
ContentType = SecretContentType.Text,
|
||||
Ciphertext = [9],
|
||||
Nonce = [9],
|
||||
Tag = [9],
|
||||
WrappedDek = [9],
|
||||
WrapNonce = [9],
|
||||
WrapTag = [9],
|
||||
KekId = "kek-out-of-band",
|
||||
Revision = 1,
|
||||
CreatedUtc = DateTimeOffset.UtcNow,
|
||||
UpdatedUtc = DateTimeOffset.UtcNow,
|
||||
};
|
||||
|
||||
public bool Contains(string name) => _rows.ContainsKey(name);
|
||||
|
||||
public Task<StoredSecret?> GetAsync(SecretName name, CancellationToken ct) =>
|
||||
Task.FromResult(_rows.TryGetValue(name.Value, out var row) ? row : null);
|
||||
|
||||
public Task UpsertAsync(StoredSecret row, CancellationToken ct)
|
||||
{
|
||||
_rows[row.Name.Value] = row;
|
||||
return Task.CompletedTask;
|
||||
}
|
||||
|
||||
public Task<bool> DeleteAsync(SecretName name, string? actor, CancellationToken ct) =>
|
||||
Task.FromResult(_rows.Remove(name.Value));
|
||||
|
||||
public Task<IReadOnlyList<SecretMetadata>> ListAsync(bool includeDeleted, CancellationToken ct) =>
|
||||
Task.FromResult<IReadOnlyList<SecretMetadata>>([]);
|
||||
|
||||
public Task<IReadOnlyList<SecretManifestEntry>> GetManifestAsync(CancellationToken ct) =>
|
||||
Task.FromResult<IReadOnlyList<SecretManifestEntry>>([]);
|
||||
|
||||
public Task ApplyReplicatedAsync(StoredSecret row, CancellationToken ct)
|
||||
{
|
||||
if (ThrowOnApply)
|
||||
{
|
||||
throw new InvalidOperationException("simulated store fault");
|
||||
}
|
||||
|
||||
// Test double only — real store applies last-writer-wins; here every apply
|
||||
// just overwrites, which is fine since these tests never seed a conflicting
|
||||
// higher local revision through this path.
|
||||
_rows[row.Name.Value] = row;
|
||||
return Task.CompletedTask;
|
||||
}
|
||||
|
||||
public Task<bool> ApplyRewrapAsync(
|
||||
StoredSecret rewrappedRow, byte[] expectedCurrentWrappedDek, CancellationToken ct) =>
|
||||
Task.FromResult(false);
|
||||
}
|
||||
|
||||
/// <summary>In-memory <see cref="ISecretResolver"/> whose per-name resolution outcome tests control.</summary>
|
||||
private sealed class FakeSecretResolver : ISecretResolver
|
||||
{
|
||||
private readonly HashSet<string> _throwing = new(StringComparer.Ordinal);
|
||||
private readonly HashSet<string> _nullReturning = new(StringComparer.Ordinal);
|
||||
|
||||
public int CallCount;
|
||||
|
||||
public void FailFor(string name) => _nullReturning.Add(name);
|
||||
public void ReturnNullFor(string name) => _nullReturning.Add(name);
|
||||
public void ThrowFor(string name) => _throwing.Add(name);
|
||||
public void ClearFailures()
|
||||
{
|
||||
_nullReturning.Clear();
|
||||
_throwing.Clear();
|
||||
}
|
||||
|
||||
public Task<string?> GetAsync(SecretName name, CancellationToken ct)
|
||||
{
|
||||
Interlocked.Increment(ref CallCount);
|
||||
if (_throwing.Contains(name.Value))
|
||||
{
|
||||
throw new InvalidOperationException("simulated decrypt failure");
|
||||
}
|
||||
|
||||
return Task.FromResult(_nullReturning.Contains(name.Value) ? null : "decrypted-value");
|
||||
}
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user