fix(transport): enforce the documented per-IP bundle-unlock throttle (row 44)

BundleUnlockRateLimiter was registered (ServiceCollectionExtensions), injected
into BundleImporter, unit-tested, documented on
TransportOptions.MaxUnlockAttemptsPerIpPerHour = 10 and pinned in
appsettings.Central.json — but _unlockRateLimiter was never invoked. A
brute-force passphrase attacker against a stolen bundle was bounded only by the
per-bundle 3-strike lockout, which a re-export side-steps (new ContentHash).

LoadAsync now registers every unlock attempt against the limiter. Ordering is
load-bearing: AFTER the per-bundle lockout check (hammering a locked bundle
spends no hourly budget) and BEFORE the decrypt (so a denial burns no per-bundle
strike — otherwise the throttle would be a way to lock a bundle someone is
legitimately importing). Over budget throws the new
BundleUnlockThrottledException, fail-closed with a clear message on both
surfaces; it lives in Commons because the management surface (Commons-only
references) must catch it to return a curated error rather than MapFault's
generic internal-error text.

Client key plumbing:
- Central UI: new scoped CircuitClientAddressProvider snapshots
  Connection.RemoteIpAddress at circuit establishment — the same constraint and
  the same pattern as CookieAuthenticationStateProvider, since HttpContext is
  null by the time a circuit submits a passphrase.
- CLI/management: additive ManagementEnvelope.ClientIp, stamped by
  ManagementEndpoints and threaded through DispatchCommand to the preview/import
  bundle handlers.
- No attributable address => a single SHARED "<unattributed>" bucket, not a
  bypass: unknown clients get a stricter budget, never none. Per-client
  isolation is only real behind a ForwardedHeaders-trusted proxy, the same
  caveat the login throttle carries (Component-Security.md).

Docs: Component-Transport.md + docs/components/Transport.md record the two-level
ordering, the client-key sourcing, the shared-bucket fallback and a
troubleshooting entry for the new error.

Tests: 6 importer tests (cap enforced; denial != passphrase failure, no strike
burned; per-client isolation; shared unattributed bucket; unencrypted bundles
consume no budget; a locked bundle spends no hourly budget) — 4 of the 6 fail
with the limiter call disabled, i.e. against the dead-code state. Plus
ManagementActor tests pinning the ClientIp forwarding and the curated
throttle error.
This commit is contained in:
Joseph Doherty
2026-08-15 06:01:14 -04:00
parent 708389829b
commit 5ab26a4233
16 changed files with 545 additions and 24 deletions
@@ -9,6 +9,7 @@ using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Options;
using NSubstitute;
using NSubstitute.ExceptionExtensions;
using ZB.MOM.WW.ScadaBridge.CentralUI.Auth;
using ZB.MOM.WW.ScadaBridge.Commons.Entities.Sites;
using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Repositories;
using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Services;
@@ -45,6 +46,9 @@ public class TransportImportPageTests : BunitContext
// the Map tests) can render the wizard without a missing-service failure.
private readonly ISiteRepository _siteRepo = Substitute.For<ISiteRepository>();
/// <summary>Client address the wizard reports for the per-IP unlock throttle (register row 44).</summary>
private const string TestClientIp = "203.0.113.7";
public TransportImportPageTests()
{
JSInterop.Mode = JSRuntimeMode.Loose;
@@ -52,6 +56,11 @@ public class TransportImportPageTests : BunitContext
Services.AddSingleton(_importer);
Services.AddSingleton(_auditService);
// Register row 44: the wizard passes the circuit's client IP to LoadAsync so
// the per-IP unlock throttle can attribute the attempt. bUnit has no HTTP
// context, so use the provider's explicit-address test seam.
Services.AddSingleton(new CircuitClientAddressProvider(TestClientIp));
_siteRepo.GetAllSitesAsync(Arg.Any<CancellationToken>())
.Returns(Array.Empty<Site>());
_siteRepo.GetDataConnectionsBySiteIdAsync(Arg.Any<int>(), Arg.Any<CancellationToken>())
@@ -132,7 +141,7 @@ public class TransportImportPageTests : BunitContext
public async Task Decryption_failure_increments_attempt_counter()
{
// Set up the importer to throw CryptographicException for wrong passphrases.
_importer.LoadAsync(Arg.Any<Stream>(), Arg.Any<string>(), Arg.Any<CancellationToken>())
_importer.LoadAsync(Arg.Any<Stream>(), Arg.Any<string>(), Arg.Any<CancellationToken>(), Arg.Any<string?>())
.Throws(new CryptographicException("authentication tag mismatch"));
var cut = Render<TransportImportPage>();
@@ -162,7 +171,7 @@ public class TransportImportPageTests : BunitContext
[Fact]
public async Task Three_failed_unlocks_force_reupload()
{
_importer.LoadAsync(Arg.Any<Stream>(), Arg.Any<string>(), Arg.Any<CancellationToken>())
_importer.LoadAsync(Arg.Any<Stream>(), Arg.Any<string>(), Arg.Any<CancellationToken>(), Arg.Any<string?>())
.Throws(new CryptographicException("authentication tag mismatch"));
var cut = Render<TransportImportPage>();
@@ -725,7 +734,8 @@ public class TransportImportPageTests : BunitContext
return Store.Open(session);
}
public Task<BundleSession> LoadAsync(Stream bundleStream, string? passphrase, CancellationToken ct = default)
public Task<BundleSession> LoadAsync(
Stream bundleStream, string? passphrase, CancellationToken ct = default, string? clientKey = null)
=> throw new NotSupportedException("These tests seed sessions via OpenSession().");
public Task<ImportPreview> PreviewAsync(Guid sessionId, CancellationToken ct = default)
@@ -234,6 +234,10 @@ public sealed class QueryStringDrillInTests
Services.AddSingleton(Substitute.For<IAuditService>());
// M8 E2: TransportImport's Map step injects ISiteRepository.
Services.AddSingleton(Substitute.For<ISiteRepository>());
// Register row 44: the wizard reads the circuit's client IP for the
// per-IP unlock throttle. No HTTP context under bUnit — use the seam.
Services.AddSingleton(new ZB.MOM.WW.ScadaBridge.CentralUI.Auth.CircuitClientAddressProvider(
clientAddress: null));
Services.AddSingleton<IOptions<TransportOptions>>(
Microsoft.Extensions.Options.Options.Create(new TransportOptions
{
@@ -2734,7 +2734,7 @@ public class ManagementActorTests : TestKit, IDisposable
var (_, importer) = AddBundleSubstitutes();
var sessionId = Guid.NewGuid();
importer.LoadAsync(Arg.Any<Stream>(), Arg.Any<string?>(), Arg.Any<CancellationToken>())
importer.LoadAsync(Arg.Any<Stream>(), Arg.Any<string?>(), Arg.Any<CancellationToken>(), Arg.Any<string?>())
.Returns(new Commons.Types.Transport.BundleSession
{
SessionId = sessionId,
@@ -2787,7 +2787,7 @@ public class ManagementActorTests : TestKit, IDisposable
var (_, importer) = AddBundleSubstitutes();
var sessionId = Guid.NewGuid();
importer.LoadAsync(Arg.Any<Stream>(), Arg.Any<string?>(), Arg.Any<CancellationToken>())
importer.LoadAsync(Arg.Any<Stream>(), Arg.Any<string?>(), Arg.Any<CancellationToken>(), Arg.Any<string?>())
.Returns(new Commons.Types.Transport.BundleSession
{
SessionId = sessionId,
@@ -2865,7 +2865,7 @@ public class ManagementActorTests : TestKit, IDisposable
IReadOnlyList<Commons.Types.Transport.RequiredConnectionMapping> connectionMappings)
{
var sessionId = Guid.NewGuid();
importer.LoadAsync(Arg.Any<Stream>(), Arg.Any<string?>(), Arg.Any<CancellationToken>())
importer.LoadAsync(Arg.Any<Stream>(), Arg.Any<string?>(), Arg.Any<CancellationToken>(), Arg.Any<string?>())
.Returns(new Commons.Types.Transport.BundleSession
{
SessionId = sessionId,
@@ -3089,7 +3089,7 @@ public class ManagementActorTests : TestKit, IDisposable
var (_, importer) = AddBundleSubstitutes();
var sessionId = Guid.NewGuid();
importer.LoadAsync(Arg.Any<Stream>(), Arg.Any<string?>(), Arg.Any<CancellationToken>())
importer.LoadAsync(Arg.Any<Stream>(), Arg.Any<string?>(), Arg.Any<CancellationToken>(), Arg.Any<string?>())
.Returns(new Commons.Types.Transport.BundleSession
{
SessionId = sessionId,
@@ -3126,6 +3126,99 @@ public class ManagementActorTests : TestKit, IDisposable
Assert.Contains("OpcA", success.JsonData);
}
[Fact]
public void PreviewBundleCommand_ReleasesTheSession_AndForwardsTheCallersIp()
{
// Register row 43: a preview never applies, so nothing else releases the
// session — it used to hold one of MaxConcurrentImportSessions for the full
// TTL. Register row 44: the envelope's client IP must reach LoadAsync so the
// per-IP-per-hour unlock throttle can attribute the attempt.
var (_, importer) = AddBundleSubstitutes();
var sessionId = Guid.NewGuid();
importer.LoadAsync(Arg.Any<Stream>(), Arg.Any<string?>(), Arg.Any<CancellationToken>(), Arg.Any<string?>())
.Returns(new Commons.Types.Transport.BundleSession
{
SessionId = sessionId,
Manifest = null!,
ExpiresAt = DateTimeOffset.UtcNow.AddMinutes(5),
});
importer.PreviewAsync(sessionId, Arg.Any<CancellationToken>())
.Returns(new Commons.Types.Transport.ImportPreview(
sessionId,
Array.Empty<Commons.Types.Transport.ImportPreviewItem>()));
var payload = Convert.ToBase64String(new byte[] { 0x01, 0x02, 0x03 });
var actor = CreateActor();
var envelope = Envelope(new PreviewBundleCommand(payload, null), "Administrator")
with { ClientIp = "198.51.100.9" };
actor.Tell(envelope);
ExpectMsg<ManagementSuccess>(TimeSpan.FromSeconds(5));
importer.Received(1).LoadAsync(
Arg.Any<Stream>(), Arg.Any<string?>(), Arg.Any<CancellationToken>(), "198.51.100.9");
importer.Received(1).Abandon(sessionId);
}
[Fact]
public void ImportBundleCommand_WithBlockerRow_StillReleasesTheSession()
{
// Register row 43: ApplyAsync releases the session, but this import aborts
// BEFORE apply — the handler must release it anyway.
var (_, importer) = AddBundleSubstitutes();
var sessionId = Guid.NewGuid();
importer.LoadAsync(Arg.Any<Stream>(), Arg.Any<string?>(), Arg.Any<CancellationToken>(), Arg.Any<string?>())
.Returns(new Commons.Types.Transport.BundleSession
{
SessionId = sessionId,
Manifest = null!,
ExpiresAt = DateTimeOffset.UtcNow.AddMinutes(5),
});
importer.PreviewAsync(sessionId, Arg.Any<CancellationToken>())
.Returns(new Commons.Types.Transport.ImportPreview(
sessionId,
new[]
{
new Commons.Types.Transport.ImportPreviewItem(
EntityType: "Template",
Name: "BlockedTemplate",
ExistingVersion: null,
IncomingVersion: 1,
Kind: Commons.Types.Transport.ConflictKind.Blocker,
FieldDiffJson: null,
BlockerReason: "FK to missing site"),
}));
var payload = Convert.ToBase64String(new byte[] { 0x01, 0x02, 0x03 });
var actor = CreateActor();
actor.Tell(Envelope(new ImportBundleCommand(payload, null, "skip"), "Administrator"));
ExpectMsg<ManagementError>(TimeSpan.FromSeconds(5));
importer.Received(1).Abandon(sessionId);
}
[Fact]
public void ImportBundleCommand_ThrottledUnlock_SurfacesTheCuratedReason()
{
// Register row 44: a per-IP throttle denial must fail closed with a clear
// message, not MapFault's generic "an internal error occurred".
var (_, importer) = AddBundleSubstitutes();
importer.LoadAsync(Arg.Any<Stream>(), Arg.Any<string?>(), Arg.Any<CancellationToken>(), Arg.Any<string?>())
.Returns<Commons.Types.Transport.BundleSession>(_ =>
throw new Commons.Types.Transport.BundleUnlockThrottledException("198.51.100.9", 10));
var payload = Convert.ToBase64String(new byte[] { 0x01, 0x02, 0x03 });
var actor = CreateActor();
actor.Tell(Envelope(new ImportBundleCommand(payload, "pass", "skip"), "Administrator"));
var response = ExpectMsg<ManagementError>(TimeSpan.FromSeconds(5));
Assert.Equal("COMMAND_FAILED", response.ErrorCode);
Assert.Contains("Too many bundle unlock attempts", response.Error);
}
// ========================================================================
// Native alarm source CRUD (Task 21)
// ========================================================================
@@ -53,7 +53,7 @@ public class TransportGateTests : TestKit, IDisposable
var loadCalls = 0;
var session = new BundleSession { SessionId = Guid.NewGuid() };
_importer.LoadAsync(Arg.Any<Stream>(), Arg.Any<string?>(), Arg.Any<CancellationToken>())
_importer.LoadAsync(Arg.Any<Stream>(), Arg.Any<string?>(), Arg.Any<CancellationToken>(), Arg.Any<string?>())
.Returns(_ =>
{
Interlocked.Increment(ref loadCalls);
@@ -502,6 +502,183 @@ public sealed class BundleImporterLoadTests
() => rig.Importer.LoadAsync(stream, passphrase: null));
}
// ─────────────────────────────────────────────────────────────────────
// Register row 44 — the per-IP-per-hour unlock throttle is actually enforced.
// BundleUnlockRateLimiter was registered, injected and unit-tested but never
// invoked, so these all passed-by-doing-nothing before the wiring.
// ─────────────────────────────────────────────────────────────────────
[Fact]
public async Task LoadAsync_rejects_unlock_attempts_past_the_per_ip_hourly_cap()
{
// Per-bundle strikes are effectively disabled here (100) so the throttle is
// unambiguously what stops the run.
var rig = BuildRig(opts =>
{
opts.MaxUnlockAttemptsPerIpPerHour = 3;
opts.MaxUnlockAttemptsPerSession = 100;
});
using var packed = PackEncryptedBundle(
rig.Serializer, rig.ManifestBuilder, rig.Encryptor, EmptyContent(), "correct");
var bundleBytes = ((MemoryStream)packed).ToArray();
for (var attempt = 1; attempt <= 3; attempt++)
{
await Assert.ThrowsAnyAsync<CryptographicException>(
() => rig.Importer.LoadAsync(
new MemoryStream(bundleBytes), passphrase: "wrong", clientKey: "10.1.1.5"));
}
var throttled = await Assert.ThrowsAsync<BundleUnlockThrottledException>(
() => rig.Importer.LoadAsync(
new MemoryStream(bundleBytes), passphrase: "wrong", clientKey: "10.1.1.5"));
Assert.Equal(3, throttled.MaxAttemptsPerWindow);
Assert.Equal("10.1.1.5", throttled.ClientKey);
// Fail closed: even the CORRECT passphrase is refused while the client is
// over budget.
await Assert.ThrowsAsync<BundleUnlockThrottledException>(
() => rig.Importer.LoadAsync(
new MemoryStream(bundleBytes), passphrase: "correct", clientKey: "10.1.1.5"));
}
[Fact]
public async Task Throttled_unlock_is_not_a_passphrase_failure_and_burns_no_strike()
{
// A throttled attempt never reaches the decryptor, so it must not consume one
// of the bundle's three strikes — otherwise the per-IP throttle would become a
// way to lock a bundle that an operator is legitimately importing.
var rig = BuildRig(opts =>
{
opts.MaxUnlockAttemptsPerIpPerHour = 1;
opts.MaxUnlockAttemptsPerSession = 3;
});
using var packed = PackEncryptedBundle(
rig.Serializer, rig.ManifestBuilder, rig.Encryptor, EmptyContent(), "correct");
var bundleBytes = ((MemoryStream)packed).ToArray();
var contentHash = rig.Serializer.ReadManifest(new MemoryStream(bundleBytes)).ContentHash;
await Assert.ThrowsAnyAsync<CryptographicException>(
() => rig.Importer.LoadAsync(
new MemoryStream(bundleBytes), passphrase: "wrong", clientKey: "10.1.1.6"));
Assert.Equal(1, rig.SessionStore.GetUnlockFailureCount(contentHash));
await Assert.ThrowsAsync<BundleUnlockThrottledException>(
() => rig.Importer.LoadAsync(
new MemoryStream(bundleBytes), passphrase: "wrong", clientKey: "10.1.1.6"));
// Still 1 — the denial burned no strike, so the bundle is not locked.
Assert.Equal(1, rig.SessionStore.GetUnlockFailureCount(contentHash));
}
[Fact]
public async Task Unlock_throttle_is_isolated_per_client()
{
// One noisy client must not spend another operator's budget.
var rig = BuildRig(opts =>
{
opts.MaxUnlockAttemptsPerIpPerHour = 1;
opts.MaxUnlockAttemptsPerSession = 100;
});
using var packed = PackEncryptedBundle(
rig.Serializer, rig.ManifestBuilder, rig.Encryptor, EmptyContent(), "correct");
var bundleBytes = ((MemoryStream)packed).ToArray();
await Assert.ThrowsAnyAsync<CryptographicException>(
() => rig.Importer.LoadAsync(
new MemoryStream(bundleBytes), passphrase: "wrong", clientKey: "10.1.1.7"));
await Assert.ThrowsAsync<BundleUnlockThrottledException>(
() => rig.Importer.LoadAsync(
new MemoryStream(bundleBytes), passphrase: "wrong", clientKey: "10.1.1.7"));
// A different client still has its full budget — reaches the decryptor and
// fails on the passphrase, not on the throttle.
await Assert.ThrowsAnyAsync<CryptographicException>(
() => rig.Importer.LoadAsync(
new MemoryStream(bundleBytes), passphrase: "wrong", clientKey: "10.1.1.8"));
}
[Fact]
public async Task Unattributed_callers_share_one_throttle_bucket_rather_than_bypassing_it()
{
// Conservative fallback: a caller with no IP is counted against a single
// shared bucket, so "unknown client" is a stricter budget, never none.
var rig = BuildRig(opts =>
{
opts.MaxUnlockAttemptsPerIpPerHour = 1;
opts.MaxUnlockAttemptsPerSession = 100;
});
using var packed = PackEncryptedBundle(
rig.Serializer, rig.ManifestBuilder, rig.Encryptor, EmptyContent(), "correct");
var bundleBytes = ((MemoryStream)packed).ToArray();
await Assert.ThrowsAnyAsync<CryptographicException>(
() => rig.Importer.LoadAsync(new MemoryStream(bundleBytes), passphrase: "wrong"));
// Blank is treated the same as null — both land in the shared bucket, which
// the first attempt already emptied.
var throttled = await Assert.ThrowsAsync<BundleUnlockThrottledException>(
() => rig.Importer.LoadAsync(
new MemoryStream(bundleBytes), passphrase: "wrong", clientKey: " "));
Assert.Equal(BundleImporter.UnattributedClientKey, throttled.ClientKey);
}
[Fact]
public async Task Unencrypted_bundles_do_not_consume_the_unlock_budget()
{
// The throttle guards passphrase brute-forcing; a plain bundle has no unlock
// step, so repeated loads must not be able to exhaust an operator's budget.
var rig = BuildRig(opts => opts.MaxUnlockAttemptsPerIpPerHour = 1);
using var packed = PackPlainBundle(rig.Serializer, rig.ManifestBuilder, SmallContent());
var bundleBytes = ((MemoryStream)packed).ToArray();
for (var i = 0; i < 3; i++)
{
var session = await rig.Importer.LoadAsync(
new MemoryStream(bundleBytes), passphrase: null, clientKey: "10.1.1.9");
Assert.NotNull(session);
rig.Importer.Abandon(session.SessionId);
}
}
[Fact]
public async Task An_already_locked_bundle_does_not_spend_the_clients_hourly_budget()
{
// Ordering pin: the per-bundle lockout is checked BEFORE the throttle, so
// hammering a locked bundle cannot burn through the client's hourly budget
// and spill over into denying its next legitimate import.
var rig = BuildRig(opts =>
{
opts.MaxUnlockAttemptsPerSession = 1;
opts.MaxUnlockAttemptsPerIpPerHour = 2;
});
using var packed = PackEncryptedBundle(
rig.Serializer, rig.ManifestBuilder, rig.Encryptor, EmptyContent(), "correct");
var bundleBytes = ((MemoryStream)packed).ToArray();
// One wrong attempt locks the bundle (cap 1) and spends 1 of 2 IP budget.
await Assert.ThrowsAsync<BundleLockedException>(
() => rig.Importer.LoadAsync(
new MemoryStream(bundleBytes), passphrase: "wrong", clientKey: "10.1.1.10"));
// Three more attempts on the locked bundle stay BundleLocked — none of them
// register against the rate limiter.
for (var i = 0; i < 3; i++)
{
await Assert.ThrowsAsync<BundleLockedException>(
() => rig.Importer.LoadAsync(
new MemoryStream(bundleBytes), passphrase: "wrong", clientKey: "10.1.1.10"));
}
// A DIFFERENT bundle from the same client still has budget left.
using var other = PackEncryptedBundle(
rig.Serializer, rig.ManifestBuilder, rig.Encryptor, SmallContent(), "correct2");
var otherSession = await rig.Importer.LoadAsync(
new MemoryStream(((MemoryStream)other).ToArray()),
passphrase: "correct2", clientKey: "10.1.1.10");
Assert.NotNull(otherSession);
}
// ─────────────────────────────────────────────────────────────────────
// Register row 43 — abandoned-session release.
// ─────────────────────────────────────────────────────────────────────