diff --git a/src/Backend/AutopilotMonitor.Functions.Tests/SoftwareKeysJsonCodecTests.cs b/src/Backend/AutopilotMonitor.Functions.Tests/SoftwareKeysJsonCodecTests.cs new file mode 100644 index 00000000..b16813ad --- /dev/null +++ b/src/Backend/AutopilotMonitor.Functions.Tests/SoftwareKeysJsonCodecTests.cs @@ -0,0 +1,223 @@ +using System.Collections.Generic; +using System.Linq; +using AutopilotMonitor.Shared.Models.Deletion; +using Xunit; + +namespace AutopilotMonitor.Functions.Tests; + +/// +/// Pure-logic tests for the SoftwareKeysJson codec used by the inventory contributions side-row +/// (Plan §17). Encoding determinism + delta computation are the load-bearing properties — the +/// cascade decrement step reads back the side-row's keys and is byte-identical to what the +/// increment path wrote. +/// +public class SoftwareKeysJsonCodecTests +{ + [Fact] + public void Encode_uses_raw_json_below_compression_threshold() + { + var keys = new List + { + new DeletionDecrementKey { Vendor = "Contoso", Name = "Widget", Version = "1.0" }, + new DeletionDecrementKey { Vendor = "Fabrikam", Name = "Gadget", Version = "2.0" }, + }; + + var encoded = SoftwareKeysJsonCodec.Encode(keys); + + Assert.False(encoded.IsCompressed); + Assert.Equal(2, encoded.KeyCount); + Assert.Contains("Contoso", encoded.Encoded); + Assert.Contains("Fabrikam", encoded.Encoded); + } + + [Fact] + public void Encode_uses_gzip_base64_above_compression_threshold() + { + // Generate >30 KB of raw JSON: 1000 distinct keys with ~30-char vendor each. + var keys = new List(); + for (var i = 0; i < 1000; i++) + { + keys.Add(new DeletionDecrementKey + { + Vendor = $"VendorWithALongerNameForBulk-{i:D6}", + Name = $"Product-{i:D6}", + Version = $"1.{i}.0", + }); + } + + var encoded = SoftwareKeysJsonCodec.Encode(keys); + + Assert.True(encoded.IsCompressed); + Assert.Equal(1000, encoded.KeyCount); + // Compressed payload should be much smaller than raw (heavy redundancy). + var rawSize = System.Text.Json.JsonSerializer.SerializeToUtf8Bytes(keys, DeletionManifestJson.SerializerOptions).Length; + Assert.True(encoded.Encoded.Length * 3 / 4 < rawSize / 3, + $"Compression should shrink payload by >3x. Got base64 {encoded.Encoded.Length} chars vs raw {rawSize} bytes."); + } + + [Fact] + public void Encode_decode_round_trips_keys_in_deterministic_order() + { + var keys = new List + { + new DeletionDecrementKey { Vendor = "Tailspin", Name = "Service", Version = "3.0" }, + new DeletionDecrementKey { Vendor = "Contoso", Name = "Widget", Version = "1.0" }, + new DeletionDecrementKey { Vendor = "Fabrikam", Name = "Gadget", Version = "2.0" }, + }; + + var encoded = SoftwareKeysJsonCodec.Encode(keys); + var decoded = SoftwareKeysJsonCodec.Decode(encoded.Encoded, encoded.IsCompressed); + + Assert.Equal(3, decoded.Count); + // Builder sorts by (Vendor, Name, Version) so the side-row content is byte-stable + // across builds — critical for manifest-determinism (Plan §18.6). + Assert.Equal("Contoso", decoded[0].Vendor); + Assert.Equal("Fabrikam", decoded[1].Vendor); + Assert.Equal("Tailspin", decoded[2].Vendor); + } + + [Fact] + public void Encode_with_compression_round_trips_through_gzip_base64() + { + // Force the compressed path with a key set that just barely exceeds the threshold. + var keys = new List(); + for (var i = 0; i < 500; i++) + { + keys.Add(new DeletionDecrementKey + { + Vendor = "VendorWithALongerNameForBulk-" + i.ToString("D6"), + Name = "Product-" + i.ToString("D6"), + Version = "1." + i + ".0", + }); + } + + var encoded = SoftwareKeysJsonCodec.Encode(keys); + var decoded = SoftwareKeysJsonCodec.Decode(encoded.Encoded, encoded.IsCompressed); + + Assert.True(encoded.IsCompressed); + Assert.Equal(500, decoded.Count); + Assert.Equal(keys[0].Vendor, decoded.First(k => k.Vendor == keys[0].Vendor).Vendor); + } + + [Fact] + public void Decode_empty_string_returns_empty_list() + { + Assert.Empty(SoftwareKeysJsonCodec.Decode("", false)); + Assert.Empty(SoftwareKeysJsonCodec.Decode("", true)); + } + + [Fact] + public void Decode_invalid_base64_throws_invalid_data() + { + var ex = Assert.Throws( + () => SoftwareKeysJsonCodec.Decode("!!! not base64 !!!", isCompressed: true)); + Assert.Contains("Base64", ex.Message); + } + + [Fact] + public void Decode_malformed_json_throws_invalid_data() + { + Assert.Throws( + () => SoftwareKeysJsonCodec.Decode("{not-an-array}", isCompressed: false)); + } + + [Fact] + public void ComputeDelta_returns_all_added_when_oldKeys_is_empty() + { + var newKeys = new[] + { + new DeletionDecrementKey { Vendor = "Contoso", Name = "Widget", Version = "1.0" }, + new DeletionDecrementKey { Vendor = "Fabrikam", Name = "Gadget", Version = "2.0" }, + }; + + var (added, removed) = SoftwareKeysJsonCodec.ComputeDelta(System.Array.Empty(), newKeys); + + Assert.Equal(2, added.Count); + Assert.Empty(removed); + } + + [Fact] + public void ComputeDelta_returns_one_added_when_newKeys_adds_a_phase2_app() + { + var oldKeys = new[] + { + new DeletionDecrementKey { Vendor = "Contoso", Name = "Widget", Version = "1.0" }, + }; + var newKeys = new[] + { + new DeletionDecrementKey { Vendor = "Contoso", Name = "Widget", Version = "1.0" }, + new DeletionDecrementKey { Vendor = "Fabrikam", Name = "NewApp", Version = "2.0" }, + }; + + var (added, removed) = SoftwareKeysJsonCodec.ComputeDelta(oldKeys, newKeys); + + Assert.Single(added); + Assert.Contains(SoftwareKeysJsonCodec.CompositeKey("Fabrikam", "NewApp", "2.0"), added); + Assert.Empty(removed); + } + + [Fact] + public void ComputeDelta_returns_one_removed_when_uninstall_happens_between_phases() + { + var oldKeys = new[] + { + new DeletionDecrementKey { Vendor = "Contoso", Name = "Widget", Version = "1.0" }, + new DeletionDecrementKey { Vendor = "Fabrikam", Name = "OldApp", Version = "0.9" }, + }; + var newKeys = new[] + { + new DeletionDecrementKey { Vendor = "Contoso", Name = "Widget", Version = "1.0" }, + }; + + var (added, removed) = SoftwareKeysJsonCodec.ComputeDelta(oldKeys, newKeys); + + Assert.Empty(added); + Assert.Single(removed); + Assert.Contains(SoftwareKeysJsonCodec.CompositeKey("Fabrikam", "OldApp", "0.9"), removed); + } + + [Fact] + public void ComputeDelta_is_empty_when_keys_are_identical() + { + var keys = new[] + { + new DeletionDecrementKey { Vendor = "Contoso", Name = "Widget", Version = "1.0" }, + new DeletionDecrementKey { Vendor = "Fabrikam", Name = "Gadget", Version = "2.0" }, + }; + + var (added, removed) = SoftwareKeysJsonCodec.ComputeDelta(keys, keys); + + Assert.Empty(added); + Assert.Empty(removed); + } + + [Fact] + public void ComputeDelta_is_case_insensitive() + { + // Upstream normalization is best-effort; "Contoso" and "contoso" must collapse so a case- + // flip between two correlation events on the same session doesn't trigger spurious deltas. + var oldKeys = new[] + { + new DeletionDecrementKey { Vendor = "Contoso", Name = "Widget", Version = "1.0" }, + }; + var newKeys = new[] + { + new DeletionDecrementKey { Vendor = "contoso", Name = "widget", Version = "1.0" }, + }; + + var (added, removed) = SoftwareKeysJsonCodec.ComputeDelta(oldKeys, newKeys); + + Assert.Empty(added); + Assert.Empty(removed); + } + + [Fact] + public void CompositeKey_builds_three_part_separated_value() + { + var ck = SoftwareKeysJsonCodec.CompositeKey("Contoso", "Widget", "1.0"); + Assert.NotEqual("ContosoWidget1.0", ck); // separator IS present + // Different orderings of the same string components produce different composites. + var ck2 = SoftwareKeysJsonCodec.CompositeKey("Widget", "Contoso", "1.0"); + Assert.NotEqual(ck, ck2); + } +} diff --git a/src/Backend/AutopilotMonitor.Functions.Tests/TableStorageDeletionBatchTests.cs b/src/Backend/AutopilotMonitor.Functions.Tests/TableStorageDeletionBatchTests.cs new file mode 100644 index 00000000..1697e226 --- /dev/null +++ b/src/Backend/AutopilotMonitor.Functions.Tests/TableStorageDeletionBatchTests.cs @@ -0,0 +1,241 @@ +using System; +using System.Collections.Generic; +using System.Linq; +using System.Threading; +using System.Threading.Tasks; +using Azure; +using Azure.Data.Tables; +using AutopilotMonitor.Functions.Services; +using AutopilotMonitor.Shared; +using Microsoft.Extensions.Logging.Abstractions; +using Moq; +using Xunit; + +namespace AutopilotMonitor.Functions.Tests; + +/// +/// SDK-mocked behaviour tests for . +/// This is the safety layer the PR4 cascade worker depends on for idempotent re-runs: groups by +/// PartitionKey, chunks at 100 actions per batch (Azure Tables' transaction limit), falls back +/// to per-row delete with 404-ignore when whole-batch rollback hides which row was already +/// missing. Tests use Moq against the Azure SDK's virtual surface — no Azurite, matching repo +/// convention. +/// +public class TableStorageDeletionBatchTests +{ + private const string TableName = Constants.TableNames.EventTypeIndex; + + [Fact] + public async Task Delete_returns_empty_result_for_empty_key_list() + { + var harness = new Harness(); + var result = await harness.Sut.DeleteByExactKeysInBatchesAsync(TableName, new List<(string, string)>()); + + Assert.Equal(0, result.Attempted); + Assert.Equal(0, result.DeletedNow); + Assert.Equal(0, result.AlreadyMissing); + Assert.Empty(harness.SubmittedBatches); + } + + [Fact] + public async Task Delete_groups_keys_by_partition_so_one_batch_per_pk() + { + var harness = new Harness(); + var keys = new List<(string Pk, string Rk)> + { + ("pkA", "rk1"), ("pkA", "rk2"), + ("pkB", "rk3"), ("pkB", "rk4"), ("pkB", "rk5"), + ("pkA", "rk6"), + }; + + var result = await harness.Sut.DeleteByExactKeysInBatchesAsync(TableName, keys); + + Assert.Equal(6, result.Attempted); + Assert.Equal(6, result.DeletedNow); + Assert.Equal(0, result.AlreadyMissing); + + // Two partitions → two batch transactions. Azure Tables forbids mixing PKs in one batch. + Assert.Equal(2, harness.SubmittedBatches.Count); + var batchPks = harness.SubmittedBatches + .Select(b => b.Select(a => a.Entity.PartitionKey).Distinct().Single()) + .OrderBy(p => p) + .ToList(); + Assert.Equal(new[] { "pkA", "pkB" }, batchPks); + } + + [Fact] + public async Task Delete_chunks_each_partition_into_batches_of_100() + { + // 250 rows in one PK → 3 batch transactions (100, 100, 50). + var harness = new Harness(); + var keys = Enumerable.Range(0, 250).Select(i => ("pkA", $"rk_{i:D4}")).ToList(); + + var result = await harness.Sut.DeleteByExactKeysInBatchesAsync(TableName, keys); + + Assert.Equal(250, result.Attempted); + Assert.Equal(250, result.DeletedNow); + Assert.Equal(3, harness.SubmittedBatches.Count); + Assert.Equal(100, harness.SubmittedBatches[0].Count); + Assert.Equal(100, harness.SubmittedBatches[1].Count); + Assert.Equal(50, harness.SubmittedBatches[2].Count); + } + + [Fact] + public async Task Delete_falls_back_to_per_row_on_batch_404_and_counts_all_already_missing() + { + // Idempotent re-run scenario: batch transaction fails 404 (Azure rolls back the whole + // batch on any 404 inside it). Per-row fallback finds every row already missing → result + // is everything in AlreadyMissing, nothing in DeletedNow, no throws. + var harness = new Harness + { + BatchBehavior = _ => throw new RequestFailedException(404, "ResourceNotFound"), + PerRowBehavior = _ => throw new RequestFailedException(404, "ResourceNotFound"), + }; + + var keys = Enumerable.Range(0, 5).Select(i => ("pkA", $"rk_{i}")).ToList(); + var result = await harness.Sut.DeleteByExactKeysInBatchesAsync(TableName, keys); + + Assert.Equal(5, result.Attempted); + Assert.Equal(0, result.DeletedNow); + Assert.Equal(5, result.AlreadyMissing); + Assert.Single(harness.SubmittedBatches); + Assert.Equal(5, harness.PerRowDeletes.Count); + } + + [Fact] + public async Task Delete_mixed_batch_404_fallback_counts_partial_deletes_and_misses() + { + // 5 rows; batch fails 404; per-row fallback finds 2 exist and 3 are missing. + var perRowState = new Dictionary + { + ["rk_0"] = true, ["rk_1"] = false, ["rk_2"] = true, ["rk_3"] = false, ["rk_4"] = false, + }; + var harness = new Harness + { + BatchBehavior = _ => throw new RequestFailedException(404, "ResourceNotFound"), + PerRowBehavior = rk => + { + if (perRowState.TryGetValue(rk, out var exists) && !exists) + throw new RequestFailedException(404, "ResourceNotFound"); + return new Mock().Object; + }, + }; + + var keys = perRowState.Keys.Select(rk => ("pkA", rk)).ToList(); + var result = await harness.Sut.DeleteByExactKeysInBatchesAsync(TableName, keys); + + Assert.Equal(5, result.Attempted); + Assert.Equal(2, result.DeletedNow); + Assert.Equal(3, result.AlreadyMissing); + } + + [Fact] + public async Task Delete_propagates_non_404_storage_errors() + { + // 500 / 503 / quota — real failures the cascade worker must surface so the queue retry can fire. + var harness = new Harness + { + BatchBehavior = _ => throw new RequestFailedException(500, "Internal Server Error"), + }; + + await Assert.ThrowsAsync( + () => harness.Sut.DeleteByExactKeysInBatchesAsync( + TableName, new List<(string, string)> { ("pkA", "rk_0") })); + } + + [Fact] + public async Task Delete_only_targets_keys_in_the_input_list_no_over_deletion() + { + var harness = new Harness(); + var keys = new List<(string, string)> + { + ("pkA", "rk_target_1"), ("pkA", "rk_target_2"), ("pkA", "rk_target_3"), + }; + + await harness.Sut.DeleteByExactKeysInBatchesAsync(TableName, keys); + + var deletedKeys = harness.SubmittedBatches + .SelectMany(b => b) + .Select(a => a.Entity.RowKey) + .OrderBy(k => k) + .ToList(); + Assert.Equal(new[] { "rk_target_1", "rk_target_2", "rk_target_3" }, deletedKeys); + + // Every action is a Delete (no accidental Upsert / Update). + foreach (var batch in harness.SubmittedBatches) + { + foreach (var action in batch) + { + Assert.Equal(TableTransactionActionType.Delete, action.ActionType); + } + } + } + + [Fact] + public async Task Delete_uses_eTag_All_so_concurrent_writes_dont_block_cascade() + { + var harness = new Harness(); + await harness.Sut.DeleteByExactKeysInBatchesAsync( + TableName, new List<(string, string)> { ("pkA", "rk_0") }); + + var batch = harness.SubmittedBatches.Single(); + Assert.Equal(ETag.All, batch.Single().Entity.ETag); + } + + // ---------------------------------------------------------------- Harness ---- + + /// + /// SDK-mock harness: and are + /// per-test overrides for the SubmitTransactionAsync / DeleteEntityAsync calls. Captured + /// invocations are accumulated in and . + /// + private sealed class Harness + { + public List> SubmittedBatches { get; } = new List>(); + public List<(string Pk, string Rk)> PerRowDeletes { get; } = new List<(string, string)>(); + + /// If set, called instead of the default success response on SubmitTransactionAsync. + public Func, Response>>? BatchBehavior { get; set; } + + /// If set, called instead of the default success response on per-row DeleteEntityAsync. + public Func? PerRowBehavior { get; set; } + + public TableStorageService Sut { get; } + + public Harness() + { + var mockTableClient = new Mock(); + + mockTableClient + .Setup(c => c.SubmitTransactionAsync(It.IsAny>(), It.IsAny())) + .Returns, CancellationToken>((actions, _) => + { + var snapshot = actions.ToList(); + SubmittedBatches.Add(snapshot); + if (BatchBehavior != null) + { + return Task.FromResult(BatchBehavior(snapshot)); + } + var inner = (IReadOnlyList)snapshot.Select(_ => new Mock().Object).ToList(); + return Task.FromResult(Response.FromValue(inner, new Mock().Object)); + }); + + mockTableClient + .Setup(c => c.DeleteEntityAsync(It.IsAny(), It.IsAny(), It.IsAny(), It.IsAny())) + .Returns((pk, rk, _, _) => + { + PerRowDeletes.Add((pk, rk)); + if (PerRowBehavior != null) + { + return Task.FromResult(PerRowBehavior(rk)); + } + return Task.FromResult(new Mock().Object); + }); + + var mockServiceClient = new Mock(); + mockServiceClient.Setup(s => s.GetTableClient(It.IsAny())).Returns(mockTableClient.Object); + + Sut = new TableStorageService(mockServiceClient.Object, NullLogger.Instance); + } + } +} diff --git a/src/Backend/AutopilotMonitor.Functions.Tests/TableStorageInventoryCasTests.cs b/src/Backend/AutopilotMonitor.Functions.Tests/TableStorageInventoryCasTests.cs new file mode 100644 index 00000000..8ec020d3 --- /dev/null +++ b/src/Backend/AutopilotMonitor.Functions.Tests/TableStorageInventoryCasTests.cs @@ -0,0 +1,374 @@ +using System; +using System.Collections.Generic; +using System.Threading; +using System.Threading.Tasks; +using Azure; +using Azure.Data.Tables; +using AutopilotMonitor.Functions.Services; +using AutopilotMonitor.Shared; +using AutopilotMonitor.Shared.Models.Deletion; +using Microsoft.Extensions.Logging.Abstractions; +using Moq; +using Xunit; + +namespace AutopilotMonitor.Functions.Tests; + +/// +/// SDK-mocked behaviour tests for the SoftwareInventory side-row + counter ETag-CAS helpers +/// in . These are the writer-side correctness primitives that +/// the cascade decrement step (PR4) and the new VulnerabilityCorrelationService increment path +/// (PR2) depend on. Tests exercise the actual retry loops + clamp + 404 idempotency without +/// hitting Azure, matching repo convention. +/// +public class TableStorageInventoryCasTests +{ + private const string TenantId = "11111111-1111-1111-1111-111111111111"; + private const string SessionId = "22222222-2222-2222-2222-222222222222"; + + // ============================================================ IncrementSoftwareInventoryEntryAsync ==== + + [Fact] + public async Task Increment_creates_row_with_full_metadata_when_no_row_exists() + { + // First encounter for a (tenant, key) pair → AddEntity with all the display + normalization + // metadata, SessionCount=1, FirstSeenAt/LastSeenAt populated. + var harness = new Harness(); + harness.SoftwareInventory.Setup(t => + t.GetEntityAsync(TenantId, It.IsAny(), It.IsAny>(), It.IsAny())) + .ThrowsAsync(new RequestFailedException(404, "ResourceNotFound")); + + TableEntity? addedEntity = null; + harness.SoftwareInventory.Setup(t => t.AddEntityAsync(It.IsAny(), It.IsAny())) + .Returns((e, _) => + { + addedEntity = e; + return Task.FromResult(new Mock().Object); + }); + + var item = new SoftwareInventoryItem + { + NormalizedVendor = "contoso", NormalizedName = "widget", NormalizedVersion = "1.0", + DisplayName = "Contoso Widget", Publisher = "Contoso Ltd.", + RegistrySource = "x64", NormalizationConfidence = "high", + CpeUri = "cpe:2.3:a:contoso:widget:*:*:*:*:*:*:*:*", + }; + + await harness.Sut.IncrementSoftwareInventoryEntryAsync(TenantId, item, SessionId); + + Assert.NotNull(addedEntity); + Assert.Equal(TenantId, addedEntity!.PartitionKey); + Assert.Equal(1, addedEntity.GetInt32("SessionCount")); + Assert.Equal("Contoso Widget", addedEntity.GetString("DisplayName")); + Assert.Equal("widget", addedEntity.GetString("NormalizedName")); + Assert.Equal("contoso", addedEntity.GetString("NormalizedVendor")); + Assert.Equal("1.0", addedEntity.GetString("NormalizedVersion")); + Assert.Equal("cpe:2.3:a:contoso:widget:*:*:*:*:*:*:*:*", addedEntity.GetString("CpeUri")); + Assert.Equal(SessionId, addedEntity.GetString("FirstSessionId")); + Assert.Equal(SessionId, addedEntity.GetString("LastSessionId")); + + harness.SoftwareInventory.Verify(t => t.AddEntityAsync(It.IsAny(), It.IsAny()), Times.Once); + harness.SoftwareInventory.Verify(t => t.UpdateEntityAsync(It.IsAny(), It.IsAny(), It.IsAny(), It.IsAny()), Times.Never); + } + + [Fact] + public async Task Increment_does_etag_cas_plus_one_when_row_exists() + { + var existing = new TableEntity(TenantId, "contoso:widget:1.0") + { + ["DisplayName"] = "Contoso Widget", + ["NormalizedName"] = "widget", + ["SessionCount"] = 3, + ["LastSeenAt"] = DateTime.UtcNow.AddDays(-1).ToString("o"), + ["CpeUri"] = "cpe:2.3:a:contoso:widget:*:*:*:*:*:*:*:*", + }; + existing.ETag = new ETag("0xABC"); + + var harness = new Harness(); + harness.SoftwareInventory.Setup(t => + t.GetEntityAsync(TenantId, It.IsAny(), It.IsAny>(), It.IsAny())) + .ReturnsAsync(Response.FromValue(existing, new Mock().Object)); + + TableEntity? written = null; + harness.SoftwareInventory.Setup(t => t.UpdateEntityAsync( + It.IsAny(), It.IsAny(), It.IsAny(), It.IsAny())) + .Returns((e, etag, _, _) => + { + written = e; + Assert.Equal(new ETag("0xABC"), etag); + return Task.FromResult(new Mock().Object); + }); + + var item = new SoftwareInventoryItem + { + NormalizedVendor = "contoso", NormalizedName = "widget", NormalizedVersion = "1.0", + }; + await harness.Sut.IncrementSoftwareInventoryEntryAsync(TenantId, item, SessionId); + + Assert.NotNull(written); + Assert.Equal(4, written!.GetInt32("SessionCount")); + Assert.Equal(SessionId, written.GetString("LastSessionId")); + harness.SoftwareInventory.Verify(t => t.AddEntityAsync(It.IsAny(), It.IsAny()), Times.Never); + } + + [Fact] + public async Task Increment_retries_on_412_etag_conflict_and_succeeds_on_second_attempt() + { + var existing = new TableEntity(TenantId, "contoso:widget:1.0") + { + ["SessionCount"] = 2, + }; + existing.ETag = new ETag("0xSTALE"); + + var harness = new Harness(); + harness.SoftwareInventory.Setup(t => + t.GetEntityAsync(TenantId, It.IsAny(), It.IsAny>(), It.IsAny())) + .ReturnsAsync(Response.FromValue(existing, new Mock().Object)); + + var updateCalls = 0; + harness.SoftwareInventory.Setup(t => t.UpdateEntityAsync( + It.IsAny(), It.IsAny(), It.IsAny(), It.IsAny())) + .Returns((e, _, _, _) => + { + updateCalls++; + if (updateCalls == 1) + { + throw new RequestFailedException(412, "Precondition Failed"); + } + return Task.FromResult(new Mock().Object); + }); + + var item = new SoftwareInventoryItem + { + NormalizedVendor = "contoso", NormalizedName = "widget", NormalizedVersion = "1.0", + }; + + await harness.Sut.IncrementSoftwareInventoryEntryAsync(TenantId, item, SessionId); + + Assert.Equal(2, updateCalls); + } + + [Fact] + public async Task Increment_retries_on_first_create_race_409_and_falls_back_to_update_path() + { + // Race: two callers see no row, both try AddEntity. First wins (no 409). Second gets 409, + // re-reads on next attempt, finds the row, and goes through the Update path. + var harness = new Harness(); + var getCalls = 0; + var existingAfterRace = new TableEntity(TenantId, "contoso:widget:1.0") + { + ["SessionCount"] = 1, + }; + existingAfterRace.ETag = new ETag("0xPOSTRACE"); + + harness.SoftwareInventory.Setup(t => + t.GetEntityAsync(TenantId, It.IsAny(), It.IsAny>(), It.IsAny())) + .Returns(() => + { + getCalls++; + if (getCalls == 1) + { + return Task.FromException>(new RequestFailedException(404, "ResourceNotFound")); + } + return Task.FromResult(Response.FromValue(existingAfterRace, new Mock().Object)); + }); + + harness.SoftwareInventory.Setup(t => t.AddEntityAsync(It.IsAny(), It.IsAny())) + .ThrowsAsync(new RequestFailedException(409, "EntityAlreadyExists")); + + var updateCalls = 0; + harness.SoftwareInventory.Setup(t => t.UpdateEntityAsync( + It.IsAny(), It.IsAny(), It.IsAny(), It.IsAny())) + .Returns((e, _, _, _) => + { + updateCalls++; + return Task.FromResult(new Mock().Object); + }); + + var item = new SoftwareInventoryItem + { + NormalizedVendor = "contoso", NormalizedName = "widget", NormalizedVersion = "1.0", + }; + await harness.Sut.IncrementSoftwareInventoryEntryAsync(TenantId, item, SessionId); + + Assert.Equal(2, getCalls); + Assert.Equal(1, updateCalls); + } + + [Fact] + public async Task Increment_skips_empty_normalized_triple_silently() + { + var harness = new Harness(); + var item = new SoftwareInventoryItem(); // all empty + await harness.Sut.IncrementSoftwareInventoryEntryAsync(TenantId, item, SessionId); + + harness.SoftwareInventory.Verify(t => + t.GetEntityAsync(It.IsAny(), It.IsAny(), It.IsAny>(), It.IsAny()), + Times.Never); + } + + // ============================================================ DecrementSoftwareInventoryEntryAsync ==== + + [Fact] + public async Task Decrement_is_no_op_when_row_missing_404() + { + // Idempotent: re-running the cascade on a session whose counters are already drained + // must not throw. The helper returns silently. + var harness = new Harness(); + harness.SoftwareInventory.Setup(t => + t.GetEntityAsync(TenantId, It.IsAny(), It.IsAny>(), It.IsAny())) + .ThrowsAsync(new RequestFailedException(404, "ResourceNotFound")); + + await harness.Sut.DecrementSoftwareInventoryEntryAsync(TenantId, "contoso", "widget", "1.0"); + + harness.SoftwareInventory.Verify(t => t.UpdateEntityAsync( + It.IsAny(), It.IsAny(), It.IsAny(), It.IsAny()), + Times.Never); + harness.SoftwareInventory.Verify(t => + t.DeleteEntityAsync(It.IsAny(), It.IsAny(), It.IsAny(), It.IsAny()), + Times.Never); + } + + [Fact] + public async Task Decrement_clamps_at_zero_and_deletes_the_row_when_count_drops_to_zero() + { + var existing = new TableEntity(TenantId, "contoso:widget:1.0") + { + ["SessionCount"] = 1, + }; + existing.ETag = new ETag("0xLASTONE"); + + var harness = new Harness(); + harness.SoftwareInventory.Setup(t => + t.GetEntityAsync(TenantId, It.IsAny(), It.IsAny>(), It.IsAny())) + .ReturnsAsync(Response.FromValue(existing, new Mock().Object)); + + harness.SoftwareInventory.Setup(t => + t.DeleteEntityAsync(TenantId, It.IsAny(), It.IsAny(), It.IsAny())) + .ReturnsAsync(new Mock().Object); + + await harness.Sut.DecrementSoftwareInventoryEntryAsync(TenantId, "contoso", "widget", "1.0"); + + // Row is deleted (count→0), NOT updated to a row with SessionCount=0. + harness.SoftwareInventory.Verify(t => + t.DeleteEntityAsync(TenantId, It.IsAny(), new ETag("0xLASTONE"), It.IsAny()), + Times.Once); + harness.SoftwareInventory.Verify(t => t.UpdateEntityAsync( + It.IsAny(), It.IsAny(), It.IsAny(), It.IsAny()), + Times.Never); + } + + [Fact] + public async Task Decrement_subtracts_one_when_count_is_greater_than_one() + { + var existing = new TableEntity(TenantId, "contoso:widget:1.0") + { + ["SessionCount"] = 4, + }; + existing.ETag = new ETag("0xSEVERAL"); + + var harness = new Harness(); + harness.SoftwareInventory.Setup(t => + t.GetEntityAsync(TenantId, It.IsAny(), It.IsAny>(), It.IsAny())) + .ReturnsAsync(Response.FromValue(existing, new Mock().Object)); + + TableEntity? written = null; + harness.SoftwareInventory.Setup(t => t.UpdateEntityAsync( + It.IsAny(), It.IsAny(), It.IsAny(), It.IsAny())) + .Returns((e, _, _, _) => + { + written = e; + return Task.FromResult(new Mock().Object); + }); + + await harness.Sut.DecrementSoftwareInventoryEntryAsync(TenantId, "contoso", "widget", "1.0"); + + Assert.NotNull(written); + Assert.Equal(3, written!.GetInt32("SessionCount")); + harness.SoftwareInventory.Verify(t => + t.DeleteEntityAsync(It.IsAny(), It.IsAny(), It.IsAny(), It.IsAny()), + Times.Never); + } + + [Fact] + public async Task Decrement_clamps_at_zero_even_when_existing_count_is_zero() + { + // Drift case: side-row promises a decrement but the counter is already at zero. Math.Max + // keeps us safe; the row is deleted (count→0 already). + var existing = new TableEntity(TenantId, "contoso:widget:1.0") + { + ["SessionCount"] = 0, + }; + existing.ETag = new ETag("0xZERO"); + + var harness = new Harness(); + harness.SoftwareInventory.Setup(t => + t.GetEntityAsync(TenantId, It.IsAny(), It.IsAny>(), It.IsAny())) + .ReturnsAsync(Response.FromValue(existing, new Mock().Object)); + + harness.SoftwareInventory.Setup(t => + t.DeleteEntityAsync(TenantId, It.IsAny(), It.IsAny(), It.IsAny())) + .ReturnsAsync(new Mock().Object); + + await harness.Sut.DecrementSoftwareInventoryEntryAsync(TenantId, "contoso", "widget", "1.0"); + + // Clamped → row deleted, never updated to a negative count. + harness.SoftwareInventory.Verify(t => + t.DeleteEntityAsync(TenantId, It.IsAny(), It.IsAny(), It.IsAny()), + Times.Once); + } + + [Fact] + public async Task Decrement_retries_on_412_and_succeeds_on_second_attempt() + { + var existing = new TableEntity(TenantId, "contoso:widget:1.0") + { + ["SessionCount"] = 5, + }; + existing.ETag = new ETag("0xSTALE"); + + var harness = new Harness(); + harness.SoftwareInventory.Setup(t => + t.GetEntityAsync(TenantId, It.IsAny(), It.IsAny>(), It.IsAny())) + .ReturnsAsync(Response.FromValue(existing, new Mock().Object)); + + var updateCalls = 0; + harness.SoftwareInventory.Setup(t => t.UpdateEntityAsync( + It.IsAny(), It.IsAny(), It.IsAny(), It.IsAny())) + .Returns((e, _, _, _) => + { + updateCalls++; + if (updateCalls == 1) + { + throw new RequestFailedException(412, "Precondition Failed"); + } + return Task.FromResult(new Mock().Object); + }); + + await harness.Sut.DecrementSoftwareInventoryEntryAsync(TenantId, "contoso", "widget", "1.0"); + + Assert.Equal(2, updateCalls); + } + + // ============================================================ Harness ==== + + /// + /// SDK-mock harness for SoftwareInventory CAS helpers. Per-test setups configure + /// for the (Get / Add / Update / Delete) call sequence + /// the inventory partial drives. + /// + private sealed class Harness + { + public Mock SoftwareInventory { get; } + public TableStorageService Sut { get; } + + public Harness() + { + SoftwareInventory = new Mock(); + var mockServiceClient = new Mock(); + mockServiceClient + .Setup(s => s.GetTableClient(Constants.TableNames.SoftwareInventory)) + .Returns(SoftwareInventory.Object); + Sut = new TableStorageService(mockServiceClient.Object, NullLogger.Instance); + } + } +} diff --git a/src/Backend/AutopilotMonitor.Functions.Tests/VulnerabilityCorrelationServiceContributionsTests.cs b/src/Backend/AutopilotMonitor.Functions.Tests/VulnerabilityCorrelationServiceContributionsTests.cs new file mode 100644 index 00000000..3e05ff0d --- /dev/null +++ b/src/Backend/AutopilotMonitor.Functions.Tests/VulnerabilityCorrelationServiceContributionsTests.cs @@ -0,0 +1,328 @@ +using System; +using System.Collections.Generic; +using System.Threading; +using System.Threading.Tasks; +using AutopilotMonitor.Functions.Services.Vulnerability; +using AutopilotMonitor.Shared.DataAccess; +using AutopilotMonitor.Shared.Models.Deletion; +using Microsoft.Extensions.Logging.Abstractions; +using Moq; +using Xunit; + +namespace AutopilotMonitor.Functions.Tests; + +/// +/// Behaviour tests for the rewritten inventory-write path +/// (). The +/// background Task.Run in CorrelateAsync just forwards to this method, so testing +/// it directly is equivalent to testing the rewritten flow end-to-end without the timing +/// uncertainty of a fire-and-forget Task. +/// +public class VulnerabilityCorrelationServiceContributionsTests +{ + private const string TenantId = "11111111-1111-1111-1111-111111111111"; + private const string SessionId = "22222222-2222-2222-2222-222222222222"; + + [Fact] + public async Task ParseSoftwareInventoryItems_extracts_canonical_normalized_triple_with_cpe() + { + var raw = new List> + { + new Dictionary + { + ["normalizedName"] = "edge", + ["normalizedPublisher"] = "microsoft", + ["normalizedVersion"] = "121.0.0.0", + ["displayName"] = "Microsoft Edge", + ["publisher"] = "Microsoft Corporation", + ["registrySource"] = "x64", + ["normalizationConfidence"] = "high", + }, + }; + var cpe = new Dictionary(StringComparer.OrdinalIgnoreCase) + { + ["edge"] = "cpe:2.3:a:microsoft:edge:*:*:*:*:*:*:*:*", + }; + + var items = VulnerabilityCorrelationService.ParseSoftwareInventoryItems(raw, cpe); + + Assert.Single(items); + Assert.Equal("microsoft", items[0].NormalizedVendor); + Assert.Equal("edge", items[0].NormalizedName); + Assert.Equal("121.0.0.0", items[0].NormalizedVersion); + Assert.Equal("Microsoft Edge", items[0].DisplayName); + Assert.Equal("cpe:2.3:a:microsoft:edge:*:*:*:*:*:*:*:*", items[0].CpeUri); + } + + [Fact] + public void ParseSoftwareInventoryItems_skips_entries_with_empty_normalized_triple() + { + var raw = new List> + { + new Dictionary { ["normalizedName"] = "good", ["normalizedPublisher"] = "pub", ["normalizedVersion"] = "1.0" }, + new Dictionary { ["displayName"] = "Empty triple — should be skipped" }, + }; + + var items = VulnerabilityCorrelationService.ParseSoftwareInventoryItems(raw, cpeMappings: null); + + Assert.Single(items); + Assert.Equal("good", items[0].NormalizedName); + } + + [Fact] + public async Task ApplyInventoryContributionsAsync_calls_increment_per_added_item_in_delta() + { + var vulnRepo = new Mock(); + vulnRepo + .Setup(r => r.UpsertSessionInventoryContributionsAsync( + TenantId, SessionId, It.IsAny>(), It.IsAny())) + .ReturnsAsync(new InventoryContributionsDelta + { + FirstTime = true, + AddedItems = new List + { + new SoftwareInventoryItem { NormalizedVendor = "v1", NormalizedName = "n1", NormalizedVersion = "1.0" }, + new SoftwareInventoryItem { NormalizedVendor = "v2", NormalizedName = "n2", NormalizedVersion = "2.0" }, + }, + RemovedKeys = new List(), + }); + + var service = NewService(vulnRepo); + var raw = new List> + { + new Dictionary { ["normalizedPublisher"] = "v1", ["normalizedName"] = "n1", ["normalizedVersion"] = "1.0" }, + new Dictionary { ["normalizedPublisher"] = "v2", ["normalizedName"] = "n2", ["normalizedVersion"] = "2.0" }, + }; + + await service.ApplyInventoryContributionsAsync(TenantId, SessionId, raw, cpeMappings: null); + + vulnRepo.Verify(r => r.IncrementSoftwareInventoryEntryAsync( + TenantId, + It.Is(i => i.NormalizedName == "n1"), + SessionId, + It.IsAny()), Times.Once); + vulnRepo.Verify(r => r.IncrementSoftwareInventoryEntryAsync( + TenantId, + It.Is(i => i.NormalizedName == "n2"), + SessionId, + It.IsAny()), Times.Once); + vulnRepo.Verify(r => r.DecrementSoftwareInventoryEntryAsync( + It.IsAny(), It.IsAny(), It.IsAny(), It.IsAny(), It.IsAny()), + Times.Never); + } + + [Fact] + public async Task ApplyInventoryContributionsAsync_calls_decrement_per_removed_key_in_delta() + { + var vulnRepo = new Mock(); + vulnRepo + .Setup(r => r.UpsertSessionInventoryContributionsAsync( + TenantId, SessionId, It.IsAny>(), It.IsAny())) + .ReturnsAsync(new InventoryContributionsDelta + { + FirstTime = false, + AddedItems = new List(), + RemovedKeys = new List + { + new DeletionDecrementKey { Vendor = "vOld", Name = "nOld", Version = "0.9" }, + }, + }); + + var service = NewService(vulnRepo); + var raw = new List> + { + new Dictionary { ["normalizedPublisher"] = "v1", ["normalizedName"] = "n1", ["normalizedVersion"] = "1.0" }, + }; + + await service.ApplyInventoryContributionsAsync(TenantId, SessionId, raw, cpeMappings: null); + + vulnRepo.Verify(r => r.DecrementSoftwareInventoryEntryAsync( + TenantId, "vOld", "nOld", "0.9", It.IsAny()), Times.Once); + vulnRepo.Verify(r => r.IncrementSoftwareInventoryEntryAsync( + It.IsAny(), It.IsAny(), It.IsAny(), It.IsAny()), + Times.Never); + } + + [Fact] + public async Task ApplyInventoryContributionsAsync_returns_without_calling_repo_for_empty_input() + { + var vulnRepo = new Mock(MockBehavior.Strict); + // Strict mock — any unexpected call fails the test. + + var service = NewService(vulnRepo); + await service.ApplyInventoryContributionsAsync(TenantId, SessionId, new List>(), null); + + vulnRepo.Verify(r => r.UpsertSessionInventoryContributionsAsync( + It.IsAny(), It.IsAny(), It.IsAny>(), It.IsAny()), + Times.Never); + } + + [Fact] + public async Task ApplyInventoryContributionsAsync_swallows_side_row_helper_exceptions_for_non_fatal_logging() + { + // Helpers are fail-loud (per memory feedback_storage_helpers_fail_soft) but the caller + // intentionally logs+swallows the side-row CAS exhaustion so the primary VulnerabilityReport + // stays stored. Drift is the documented trade-off (§17.3 "Side-row write before counter writes"). + var vulnRepo = new Mock(); + vulnRepo + .Setup(r => r.UpsertSessionInventoryContributionsAsync( + TenantId, SessionId, It.IsAny>(), It.IsAny())) + .ThrowsAsync(new InvalidOperationException("SessionInventoryContributions ETag-CAS exhausted ...")); + + var service = NewService(vulnRepo); + var raw = new List> + { + new Dictionary { ["normalizedPublisher"] = "v1", ["normalizedName"] = "n1", ["normalizedVersion"] = "1.0" }, + }; + + // Should not throw. + await service.ApplyInventoryContributionsAsync(TenantId, SessionId, raw, cpeMappings: null); + + vulnRepo.Verify(r => r.IncrementSoftwareInventoryEntryAsync( + It.IsAny(), It.IsAny(), It.IsAny(), It.IsAny()), + Times.Never); + } + + [Fact] + public async Task ApplyInventoryContributionsAsync_continues_after_per_counter_failure_and_audits_each() + { + // Codex Round 2: a single Increment failure must NOT abort the rest of the loop — + // the side-row already promises the new key set, so subsequent runs would see an empty + // delta and never retry the under-counted keys. Per Plan §17.3 + §17.8, each failure + // emits an inventory_counter_write_failed audit + we keep processing the rest. + var vulnRepo = new Mock(); + vulnRepo + .Setup(r => r.UpsertSessionInventoryContributionsAsync( + TenantId, SessionId, It.IsAny>(), It.IsAny())) + .ReturnsAsync(new InventoryContributionsDelta + { + FirstTime = true, + AddedItems = new List + { + new SoftwareInventoryItem { NormalizedVendor = "v1", NormalizedName = "n1", NormalizedVersion = "1.0" }, + new SoftwareInventoryItem { NormalizedVendor = "v2", NormalizedName = "n2", NormalizedVersion = "2.0" }, + new SoftwareInventoryItem { NormalizedVendor = "v3", NormalizedName = "n3", NormalizedVersion = "3.0" }, + }, + RemovedKeys = new List + { + new DeletionDecrementKey { Vendor = "vOld", Name = "nOld", Version = "0.9" }, + }, + }); + + // Increment of (v1,n1,1.0) succeeds, (v2,n2,2.0) THROWS, (v3,n3,3.0) succeeds. + vulnRepo.Setup(r => r.IncrementSoftwareInventoryEntryAsync( + TenantId, It.Is(i => i.NormalizedName == "n2"), + SessionId, It.IsAny())) + .ThrowsAsync(new InvalidOperationException("SoftwareInventory increment ETag-CAS exhausted")); + + var maintenanceRepo = new Mock(); + maintenanceRepo.Setup(m => m.LogAuditEntryAsync( + It.IsAny(), It.IsAny(), It.IsAny(), + It.IsAny(), It.IsAny(), It.IsAny?>())) + .ReturnsAsync(true); + + var service = NewService(vulnRepo, maintenanceRepo); + var raw = new List> + { + new Dictionary { ["normalizedPublisher"] = "v1", ["normalizedName"] = "n1", ["normalizedVersion"] = "1.0" }, + new Dictionary { ["normalizedPublisher"] = "v2", ["normalizedName"] = "n2", ["normalizedVersion"] = "2.0" }, + new Dictionary { ["normalizedPublisher"] = "v3", ["normalizedName"] = "n3", ["normalizedVersion"] = "3.0" }, + }; + + // Should NOT throw. + await service.ApplyInventoryContributionsAsync(TenantId, SessionId, raw, cpeMappings: null); + + // All three increments attempted. + vulnRepo.Verify(r => r.IncrementSoftwareInventoryEntryAsync( + TenantId, It.Is(i => i.NormalizedName == "n1"), + SessionId, It.IsAny()), Times.Once); + vulnRepo.Verify(r => r.IncrementSoftwareInventoryEntryAsync( + TenantId, It.Is(i => i.NormalizedName == "n2"), + SessionId, It.IsAny()), Times.Once); + vulnRepo.Verify(r => r.IncrementSoftwareInventoryEntryAsync( + TenantId, It.Is(i => i.NormalizedName == "n3"), + SessionId, It.IsAny()), Times.Once); + // Decrement still runs after the loop continues past the failed Increment. + vulnRepo.Verify(r => r.DecrementSoftwareInventoryEntryAsync( + TenantId, "vOld", "nOld", "0.9", It.IsAny()), Times.Once); + + // Audit emitted exactly once for the failed (v2,n2,2.0) Increment. + maintenanceRepo.Verify(m => m.LogAuditEntryAsync( + TenantId, + "inventory_counter_write_failed", + "Session", + SessionId, + "system", + It.Is?>(d => + d != null + && d["operation"] == "increment" + && d["vendor"] == "v2" + && d["name"] == "n2" + && d["version"] == "2.0")), + Times.Once); + } + + [Fact] + public async Task ApplyInventoryContributionsAsync_audits_decrement_failures_too() + { + var vulnRepo = new Mock(); + vulnRepo + .Setup(r => r.UpsertSessionInventoryContributionsAsync( + TenantId, SessionId, It.IsAny>(), It.IsAny())) + .ReturnsAsync(new InventoryContributionsDelta + { + AddedItems = new List(), + RemovedKeys = new List + { + new DeletionDecrementKey { Vendor = "vDead", Name = "nDead", Version = "0.5" }, + }, + }); + + vulnRepo.Setup(r => r.DecrementSoftwareInventoryEntryAsync( + TenantId, "vDead", "nDead", "0.5", It.IsAny())) + .ThrowsAsync(new InvalidOperationException("Decrement ETag-CAS exhausted")); + + var maintenanceRepo = new Mock(); + maintenanceRepo.Setup(m => m.LogAuditEntryAsync( + It.IsAny(), It.IsAny(), It.IsAny(), + It.IsAny(), It.IsAny(), It.IsAny?>())) + .ReturnsAsync(true); + + var service = NewService(vulnRepo, maintenanceRepo); + var raw = new List> + { + new Dictionary { ["normalizedPublisher"] = "v1", ["normalizedName"] = "n1", ["normalizedVersion"] = "1.0" }, + }; + + await service.ApplyInventoryContributionsAsync(TenantId, SessionId, raw, cpeMappings: null); + + maintenanceRepo.Verify(m => m.LogAuditEntryAsync( + TenantId, + "inventory_counter_write_failed", + "Session", + SessionId, + "system", + It.Is?>(d => + d != null + && d["operation"] == "decrement" + && d["vendor"] == "vDead" + && d["name"] == "nDead" + && d["version"] == "0.5")), + Times.Once); + } + + private static VulnerabilityCorrelationService NewService( + Mock vulnRepo, Mock? maintenanceRepo = null) + { + return new VulnerabilityCorrelationService( + vulnRepo.Object, + sessionRepo: Mock.Of(), + nvdApi: null!, + kevService: null!, + msrcCorrelation: null!, + adminConfig: null!, + telemetryClient: null, + logger: NullLogger.Instance, + maintenanceRepo: maintenanceRepo?.Object); + } +} diff --git a/src/Backend/AutopilotMonitor.Functions/DataAccess/TableStorage/TableVulnerabilityRepository.cs b/src/Backend/AutopilotMonitor.Functions/DataAccess/TableStorage/TableVulnerabilityRepository.cs index 94aa0f59..90f8660b 100644 --- a/src/Backend/AutopilotMonitor.Functions/DataAccess/TableStorage/TableVulnerabilityRepository.cs +++ b/src/Backend/AutopilotMonitor.Functions/DataAccess/TableStorage/TableVulnerabilityRepository.cs @@ -1,7 +1,9 @@ +using System.Threading; using Azure; using Azure.Data.Tables; using AutopilotMonitor.Shared; using AutopilotMonitor.Shared.DataAccess; +using AutopilotMonitor.Shared.Models.Deletion; using AutopilotMonitor.Functions.Services; using Microsoft.Extensions.Logging; @@ -69,6 +71,27 @@ namespace AutopilotMonitor.Functions.DataAccess.TableStorage return results; } + // --- Software Inventory v2 (Plan §17 contributions side-row + delta-driven counters) --- + + public Task?> GetSessionInventoryContributionsAsync( + string tenantId, string sessionId, CancellationToken ct = default) + => _storage.GetSessionInventoryContributionsAsync(tenantId, sessionId, ct); + + public Task UpsertSessionInventoryContributionsAsync( + string tenantId, string sessionId, IReadOnlyList newItems, CancellationToken ct = default) + => _storage.UpsertSessionInventoryContributionsAsync(tenantId, sessionId, newItems, ct); + + public Task DeleteSessionInventoryContributionsAsync(string tenantId, string sessionId, CancellationToken ct = default) + => _storage.DeleteSessionInventoryContributionsAsync(tenantId, sessionId, ct); + + public Task IncrementSoftwareInventoryEntryAsync( + string tenantId, SoftwareInventoryItem item, string sessionId, CancellationToken ct = default) + => _storage.IncrementSoftwareInventoryEntryAsync(tenantId, item, sessionId, ct); + + public Task DecrementSoftwareInventoryEntryAsync( + string tenantId, string normalizedVendor, string normalizedName, string normalizedVersion, CancellationToken ct = default) + => _storage.DecrementSoftwareInventoryEntryAsync(tenantId, normalizedVendor, normalizedName, normalizedVersion, ct); + public async Task UpdateSoftwareInventoryCpeAsync(string normalizedName, string cpeUri) { var tableClient = _storage.GetTableClient(Constants.TableNames.SoftwareInventory); diff --git a/src/Backend/AutopilotMonitor.Functions/Services/TableStorageService.Deletion.cs b/src/Backend/AutopilotMonitor.Functions/Services/TableStorageService.Deletion.cs index a2fd2404..adce3fe5 100644 --- a/src/Backend/AutopilotMonitor.Functions/Services/TableStorageService.Deletion.cs +++ b/src/Backend/AutopilotMonitor.Functions/Services/TableStorageService.Deletion.cs @@ -1,13 +1,39 @@ +using System; using System.Collections.Generic; +using System.Linq; using System.Runtime.CompilerServices; using System.Threading; using System.Threading.Tasks; using Azure; using Azure.Data.Tables; +using AutopilotMonitor.Functions.Security; using AutopilotMonitor.Functions.Services.Deletion; namespace AutopilotMonitor.Functions.Services { + /// + /// Result of — accurate + /// progress accounting for the cascade worker. is the input key + /// count, is the number of rows that the helper actually removed + /// in this call, is the number that returned 404 (idempotent + /// re-run of a partially-completed cascade). + /// + public class DeletionBatchResult + { + public int Attempted { get; } + public int DeletedNow { get; } + public int AlreadyMissing { get; } + + public DeletionBatchResult(int attempted, int deletedNow, int alreadyMissing) + { + Attempted = attempted; + DeletedNow = deletedNow; + AlreadyMissing = alreadyMissing; + } + + public static readonly DeletionBatchResult Empty = new DeletionBatchResult(0, 0, 0); + } + /// /// Cascade-deletion read surface. Implements /// against the existing _tableServiceClient. Pure delegation; fail-loud on every @@ -68,5 +94,167 @@ namespace AutopilotMonitor.Functions.Services return null; } } + + // ============================================================ Delete helpers (PR2) ==== + + // Azure Tables batch transactions cap at 100 actions per submission, all sharing a PartitionKey. + private const int BatchActionLimit = 100; + + /// + /// Idempotent batched deletion by exact (PK, RK) keys. Groups by PartitionKey + /// (batch transactions are partition-scoped), chunks each group into -row + /// transactions, and falls back to per-row + /// with 404-ignore when the whole-batch rollback hides which row was already missing. + /// Returns for accurate progress accounting on + /// re-runs of partially-completed cascades. + /// + public async Task DeleteByExactKeysInBatchesAsync( + string tableName, + IReadOnlyList<(string Pk, string Rk)> keys, + CancellationToken cancellationToken = default) + { + if (keys == null) throw new ArgumentNullException(nameof(keys)); + var attempted = keys.Count; + if (attempted == 0) return DeletionBatchResult.Empty; + + var tableClient = _tableServiceClient.GetTableClient(tableName); + var deletedNow = 0; + var alreadyMissing = 0; + + foreach (var group in keys.GroupBy(k => k.Pk, StringComparer.Ordinal)) + { + var rowsInGroup = group.ToList(); + for (var i = 0; i < rowsInGroup.Count; i += BatchActionLimit) + { + var chunk = rowsInGroup.Skip(i).Take(BatchActionLimit).ToList(); + var actions = chunk + .Select(k => new TableTransactionAction( + TableTransactionActionType.Delete, + new TableEntity(k.Pk, k.Rk) { ETag = ETag.All })) + .ToList(); + + try + { + await tableClient.SubmitTransactionAsync(actions, cancellationToken); + deletedNow += chunk.Count; + } + catch (RequestFailedException ex) when (ex.Status == 404) + { + // Azure rolls back the entire transaction when a single Delete returns 404. + // Fall back to per-row delete with 404-ignore so the helper stays idempotent. + foreach (var (pk, rk) in chunk) + { + cancellationToken.ThrowIfCancellationRequested(); + try + { + await tableClient.DeleteEntityAsync(pk, rk, ETag.All, cancellationToken); + deletedNow++; + } + catch (RequestFailedException rfe) when (rfe.Status == 404) + { + alreadyMissing++; + } + } + } + } + } + + return new DeletionBatchResult(attempted, deletedNow, alreadyMissing); + } + + // ---------------- DISCRIMINATOR_PK_RK_SUFFIX / DISCRIMINATOR_PK_RK_EXACT -------------- + + public Task DeleteEventTypeIndexEntriesAsync( + string tenantId, string sessionId, IReadOnlyList<(string Pk, string Rk)> keys, CancellationToken ct = default) + { + SecurityValidator.EnsureValidGuid(tenantId, nameof(tenantId)); + SecurityValidator.EnsureValidGuid(sessionId, nameof(sessionId)); + return DeleteByExactKeysInBatchesAsync(Shared.Constants.TableNames.EventTypeIndex, keys, ct); + } + + public Task DeleteCveIndexEntriesAsync( + string tenantId, string sessionId, IReadOnlyList<(string Pk, string Rk)> keys, CancellationToken ct = default) + { + SecurityValidator.EnsureValidGuid(tenantId, nameof(tenantId)); + SecurityValidator.EnsureValidGuid(sessionId, nameof(sessionId)); + return DeleteByExactKeysInBatchesAsync(Shared.Constants.TableNames.CveIndex, keys, ct); + } + + // ---------------- DISCRIMINATOR_PK_PROP ----------------------------------------------- + + public Task DeleteSessionsByTerminalAsync( + IReadOnlyList<(string Pk, string Rk)> keys, CancellationToken ct = default) + => DeleteByExactKeysInBatchesAsync(Shared.Constants.TableNames.SessionsByTerminal, keys, ct); + + public Task DeleteSessionsByStageAsync( + IReadOnlyList<(string Pk, string Rk)> keys, CancellationToken ct = default) + => DeleteByExactKeysInBatchesAsync(Shared.Constants.TableNames.SessionsByStage, keys, ct); + + public Task DeleteDeadEndsByReasonAsync( + IReadOnlyList<(string Pk, string Rk)> keys, CancellationToken ct = default) + => DeleteByExactKeysInBatchesAsync(Shared.Constants.TableNames.DeadEndsByReason, keys, ct); + + public Task DeleteClassifierVerdictsByIdLevelAsync( + IReadOnlyList<(string Pk, string Rk)> keys, CancellationToken ct = default) + => DeleteByExactKeysInBatchesAsync(Shared.Constants.TableNames.ClassifierVerdictsByIdLevel, keys, ct); + + public Task DeleteSignalsByKindAsync( + IReadOnlyList<(string Pk, string Rk)> keys, CancellationToken ct = default) + => DeleteByExactKeysInBatchesAsync(Shared.Constants.TableNames.SignalsByKind, keys, ct); + + // ---------------- PK_RK_EXACT (single-row) -------------------------------------------- + + public async Task DeleteDeviceSnapshotAsync(string tenantId, string sessionId, CancellationToken ct = default) + { + SecurityValidator.EnsureValidGuid(tenantId, nameof(tenantId)); + SecurityValidator.EnsureValidGuid(sessionId, nameof(sessionId)); + var tableClient = _tableServiceClient.GetTableClient(Shared.Constants.TableNames.DeviceSnapshot); + try + { + await tableClient.DeleteEntityAsync(tenantId, sessionId, ETag.All, ct); + } + catch (RequestFailedException ex) when (ex.Status == 404) + { + // Idempotent — already gone. + } + } + + // ---------------- PK_BY_SESSION (enumerate-then-delete) ------------------------------- + + public Task DeleteSessionSignalsAsync(string tenantId, string sessionId, CancellationToken ct = default) + => DeletePkBySessionAsync(Shared.Constants.TableNames.Signals, tenantId, sessionId, ct); + + public Task DeleteSessionDecisionTransitionsAsync(string tenantId, string sessionId, CancellationToken ct = default) + => DeletePkBySessionAsync(Shared.Constants.TableNames.DecisionTransitions, tenantId, sessionId, ct); + + /// + /// Enumerates every row in with + /// PartitionKey == {tenantId}_{sessionId} and deletes them via + /// . Used for PK_BY_SESSION-class cascade + /// targets when the caller does not yet have a manifest in hand (e.g. legacy direct-delete + /// paths or test fixtures); the §1 P6 cascade worker passes manifest keys to + /// directly. + /// + private async Task DeletePkBySessionAsync( + string tableName, string tenantId, string sessionId, CancellationToken ct) + { + SecurityValidator.EnsureValidGuid(tenantId, nameof(tenantId)); + SecurityValidator.EnsureValidGuid(sessionId, nameof(sessionId)); + + var tableClient = _tableServiceClient.GetTableClient(tableName); + var partitionKey = $"{tenantId}_{sessionId}"; + var safePk = ODataSanitizer.EscapeValue(partitionKey); + + var keys = new List<(string Pk, string Rk)>(); + await foreach (var entity in tableClient.QueryAsync( + filter: $"PartitionKey eq '{safePk}'", + select: new[] { "PartitionKey", "RowKey" }, + cancellationToken: ct)) + { + keys.Add((entity.PartitionKey, entity.RowKey)); + } + + return await DeleteByExactKeysInBatchesAsync(tableName, keys, ct); + } } } diff --git a/src/Backend/AutopilotMonitor.Functions/Services/TableStorageService.Inventory.cs b/src/Backend/AutopilotMonitor.Functions/Services/TableStorageService.Inventory.cs new file mode 100644 index 00000000..3d1e0504 --- /dev/null +++ b/src/Backend/AutopilotMonitor.Functions/Services/TableStorageService.Inventory.cs @@ -0,0 +1,380 @@ +using System; +using System.Collections.Generic; +using System.Linq; +using System.Threading; +using System.Threading.Tasks; +using Azure; +using Azure.Data.Tables; +using AutopilotMonitor.Functions.Security; +using AutopilotMonitor.Shared; +using AutopilotMonitor.Shared.Models.Deletion; + +namespace AutopilotMonitor.Functions.Services +{ + /// + /// SoftwareInventory side-row + counter helpers (Plan §17). All methods are fail-loud per + /// memory feedback_storage_helpers_fail_soft — non-404 storage errors propagate; + /// bounded retry exhaustion (10 attempts / 60s wall-clock per §12-Q10) throws + /// . + /// + public partial class TableStorageService + { + // Bounded-retry budget for every ETag-CAS / AddEntity-race loop in this file. + // §12-Q10: 10 attempts OR 60s wall-clock, whichever first → poison. + private const int InventoryCasMaxAttempts = 10; + private static readonly TimeSpan InventoryCasMaxWallClock = TimeSpan.FromSeconds(60); + + // ============================================================ SessionInventoryContributions ==== + + /// + /// Reads the side-row by (tenantId, sessionId). Returns null on 404 (pre-side-row session + /// or never written). Fail-loud on any other storage error. + /// + public async Task?> GetSessionInventoryContributionsAsync( + string tenantId, string sessionId, CancellationToken ct = default) + { + SecurityValidator.EnsureValidGuid(tenantId, nameof(tenantId)); + SecurityValidator.EnsureValidGuid(sessionId, nameof(sessionId)); + + var tableClient = _tableServiceClient.GetTableClient(Constants.TableNames.SessionInventoryContributions); + try + { + var response = await tableClient.GetEntityAsync(tenantId, sessionId, cancellationToken: ct); + var encoded = response.Value.GetString("SoftwareKeysJson") ?? string.Empty; + var compressed = response.Value.GetBoolean("IsCompressed") ?? false; + return SoftwareKeysJsonCodec.Decode(encoded, compressed); + } + catch (RequestFailedException ex) when (ex.Status == 404) + { + return null; + } + } + + /// + /// Direct delete with 404-ignore. Cascade tombstone calls this after consuming the + /// side-row's keys for the decrement step (Plan §17.5). + /// + public async Task DeleteSessionInventoryContributionsAsync( + string tenantId, string sessionId, CancellationToken ct = default) + { + SecurityValidator.EnsureValidGuid(tenantId, nameof(tenantId)); + SecurityValidator.EnsureValidGuid(sessionId, nameof(sessionId)); + + var tableClient = _tableServiceClient.GetTableClient(Constants.TableNames.SessionInventoryContributions); + try + { + await tableClient.DeleteEntityAsync(tenantId, sessionId, ETag.All, ct); + } + catch (RequestFailedException ex) when (ex.Status == 404) + { + // Already gone — idempotent. + } + } + + /// + /// At-most-once-per-session, delta-update upsert. Plan §17.3 algorithm: + /// + /// Read existing side-row. + /// Compute newKeys = items.Select(ToDecrementKey). + /// If no side-row: AddEntity (full payload). On 409 → race lost, restart loop. + /// If side-row exists: decode old keys, compute (added, removed). If both empty: no-op return. + /// Otherwise UpdateEntity with ETag-Replace. On 412 → restart loop with fresh read. + /// + /// Bounded retry exhausted → throw . + /// + public async Task UpsertSessionInventoryContributionsAsync( + string tenantId, string sessionId, + IReadOnlyList newItems, + CancellationToken ct = default) + { + SecurityValidator.EnsureValidGuid(tenantId, nameof(tenantId)); + SecurityValidator.EnsureValidGuid(sessionId, nameof(sessionId)); + if (newItems == null) throw new ArgumentNullException(nameof(newItems)); + + var tableClient = _tableServiceClient.GetTableClient(Constants.TableNames.SessionInventoryContributions); + + // Deduplicate by composite key — upstream may pass the same normalization twice. + var newKeys = new Dictionary(StringComparer.OrdinalIgnoreCase); + foreach (var item in newItems) + { + if (string.IsNullOrEmpty(item.NormalizedVendor) + && string.IsNullOrEmpty(item.NormalizedName) + && string.IsNullOrEmpty(item.NormalizedVersion)) + { + continue; // skip empty keys; matches existing UpsertSoftwareInventory dedup + } + var composite = SoftwareKeysJsonCodec.CompositeKey(item.NormalizedVendor, item.NormalizedName, item.NormalizedVersion); + newKeys[composite] = item; + } + var newKeyList = newKeys.Values.Select(i => i.ToDecrementKey()).ToList(); + + var deadline = DateTime.UtcNow + InventoryCasMaxWallClock; + for (var attempt = 0; attempt < InventoryCasMaxAttempts && DateTime.UtcNow < deadline; attempt++) + { + ct.ThrowIfCancellationRequested(); + + TableEntity? existing = null; + try + { + var response = await tableClient.GetEntityAsync(tenantId, sessionId, cancellationToken: ct); + existing = response.Value; + } + catch (RequestFailedException ex) when (ex.Status == 404) + { + // First-time write path. + } + + var encoded = SoftwareKeysJsonCodec.Encode(newKeyList); + var now = DateTime.UtcNow; + + if (existing == null) + { + var row = new TableEntity(tenantId, sessionId) + { + ["SoftwareKeysJson"] = encoded.Encoded, + ["IsCompressed"] = encoded.IsCompressed, + ["CountedAt"] = now, + ["LastUpdatedAt"] = now, + ["KeyCount"] = encoded.KeyCount, + }; + try + { + await tableClient.AddEntityAsync(row, ct); + return new InventoryContributionsDelta + { + AddedItems = newKeys.Values.ToList(), + RemovedKeys = new List(), + FirstTime = true, + }; + } + catch (RequestFailedException ex) when (ex.Status == 409) + { + continue; // race lost — another writer beat us to first-insert + } + } + else + { + var oldEncoded = existing.GetString("SoftwareKeysJson") ?? string.Empty; + var oldCompressed = existing.GetBoolean("IsCompressed") ?? false; + var oldKeys = SoftwareKeysJsonCodec.Decode(oldEncoded, oldCompressed); + + var (added, removed) = SoftwareKeysJsonCodec.ComputeDelta(oldKeys, newKeyList); + if (added.Count == 0 && removed.Count == 0) + { + return new InventoryContributionsDelta + { + AddedItems = new List(), + RemovedKeys = new List(), + FirstTime = false, + }; + } + + existing["SoftwareKeysJson"] = encoded.Encoded; + existing["IsCompressed"] = encoded.IsCompressed; + existing["LastUpdatedAt"] = now; + existing["KeyCount"] = encoded.KeyCount; + + try + { + await tableClient.UpdateEntityAsync(existing, existing.ETag, TableUpdateMode.Replace, ct); + var addedItems = newKeys + .Where(kv => added.Contains(kv.Key)) + .Select(kv => kv.Value) + .ToList(); + var removedKeys = oldKeys + .Where(k => removed.Contains(SoftwareKeysJsonCodec.CompositeKey(k))) + .ToList(); + return new InventoryContributionsDelta + { + AddedItems = addedItems, + RemovedKeys = removedKeys, + FirstTime = false, + }; + } + catch (RequestFailedException ex) when (ex.Status == 412) + { + continue; // ETag conflict — re-read on next attempt + } + } + } + + throw new InvalidOperationException( + $"SessionInventoryContributions ETag-CAS exhausted after {InventoryCasMaxAttempts} attempts or " + + $"{InventoryCasMaxWallClock.TotalSeconds:0}s wall-clock. tenant={tenantId} session={sessionId}"); + } + + // ============================================================ SoftwareInventory counters ==== + + /// + /// ETag-CAS SessionCount += 1 for the (tenantId, normalizedKey) row, creating the + /// row with full metadata on first encounter. Bounded retry per + /// / . + /// + public async Task IncrementSoftwareInventoryEntryAsync( + string tenantId, SoftwareInventoryItem item, string sessionId, CancellationToken ct = default) + { + SecurityValidator.EnsureValidGuid(tenantId, nameof(tenantId)); + if (item == null) throw new ArgumentNullException(nameof(item)); + + var rowKey = BuildSoftwareInventoryRowKey(item.NormalizedVendor, item.NormalizedName, item.NormalizedVersion); + if (string.IsNullOrWhiteSpace(rowKey) || rowKey == "::") return; // skip empty triples — matches existing UpsertSoftwareInventory dedup + + var tableClient = _tableServiceClient.GetTableClient(Constants.TableNames.SoftwareInventory); + var deadline = DateTime.UtcNow + InventoryCasMaxWallClock; + + for (var attempt = 0; attempt < InventoryCasMaxAttempts && DateTime.UtcNow < deadline; attempt++) + { + ct.ThrowIfCancellationRequested(); + var nowIso = DateTime.UtcNow.ToString("o"); + + TableEntity? existing = null; + try + { + var response = await tableClient.GetEntityAsync(tenantId, rowKey, cancellationToken: ct); + existing = response.Value; + } + catch (RequestFailedException ex) when (ex.Status == 404) + { + // First-time create path. + } + + if (existing == null) + { + var row = new TableEntity(tenantId, rowKey) + { + ["DisplayName"] = item.DisplayName, + ["NormalizedName"] = item.NormalizedName, + ["NormalizedVendor"] = item.NormalizedVendor, + ["NormalizedVersion"] = item.NormalizedVersion, + ["Publisher"] = item.Publisher, + ["RegistrySource"] = item.RegistrySource, + ["NormalizationConfidence"] = item.NormalizationConfidence, + ["FirstSeenAt"] = nowIso, + ["LastSeenAt"] = nowIso, + ["FirstSessionId"] = sessionId, + ["LastSessionId"] = sessionId, + ["SessionCount"] = 1, + ["CpeUri"] = item.CpeUri ?? string.Empty, + }; + try + { + await tableClient.AddEntityAsync(row, ct); + return; + } + catch (RequestFailedException ex) when (ex.Status == 409) + { + continue; // race lost — another writer created it; reread + increment + } + } + else + { + existing["SessionCount"] = (existing.GetInt32("SessionCount") ?? 0) + 1; + existing["LastSeenAt"] = nowIso; + existing["LastSessionId"] = sessionId; + // Stamp a CpeUri if we just got one and the row didn't have one yet — preserves + // existing behaviour from UpsertSoftwareInventoryAsync. + if (!string.IsNullOrEmpty(item.CpeUri) && string.IsNullOrEmpty(existing.GetString("CpeUri"))) + { + existing["CpeUri"] = item.CpeUri; + } + + try + { + await tableClient.UpdateEntityAsync(existing, existing.ETag, TableUpdateMode.Replace, ct); + return; + } + catch (RequestFailedException ex) when (ex.Status == 412) + { + continue; + } + } + } + + throw new InvalidOperationException( + $"SoftwareInventory increment ETag-CAS exhausted after {InventoryCasMaxAttempts} attempts or " + + $"{InventoryCasMaxWallClock.TotalSeconds:0}s wall-clock. tenant={tenantId} key={rowKey}"); + } + + /// + /// ETag-CAS SessionCount -= 1 with clamp at zero. Deletes the row when count hits + /// zero. 404 = idempotent no-op. Bounded retry per / + /// ; throws on exhaustion. + /// + public async Task DecrementSoftwareInventoryEntryAsync( + string tenantId, string normalizedVendor, string normalizedName, string normalizedVersion, + CancellationToken ct = default) + { + SecurityValidator.EnsureValidGuid(tenantId, nameof(tenantId)); + var rowKey = BuildSoftwareInventoryRowKey(normalizedVendor, normalizedName, normalizedVersion); + if (string.IsNullOrWhiteSpace(rowKey) || rowKey == "::") return; + + var tableClient = _tableServiceClient.GetTableClient(Constants.TableNames.SoftwareInventory); + var deadline = DateTime.UtcNow + InventoryCasMaxWallClock; + + for (var attempt = 0; attempt < InventoryCasMaxAttempts && DateTime.UtcNow < deadline; attempt++) + { + ct.ThrowIfCancellationRequested(); + + TableEntity existing; + try + { + var response = await tableClient.GetEntityAsync(tenantId, rowKey, cancellationToken: ct); + existing = response.Value; + } + catch (RequestFailedException ex) when (ex.Status == 404) + { + return; // already gone — decrement is idempotent + } + + var newCount = Math.Max(0, (existing.GetInt32("SessionCount") ?? 0) - 1); + if (newCount == 0) + { + try + { + await tableClient.DeleteEntityAsync(tenantId, rowKey, existing.ETag, ct); + return; + } + catch (RequestFailedException ex) when (ex.Status == 412) + { + continue; + } + catch (RequestFailedException ex) when (ex.Status == 404) + { + return; + } + } + + existing["SessionCount"] = newCount; + existing["LastSeenAt"] = DateTime.UtcNow.ToString("o"); + try + { + await tableClient.UpdateEntityAsync(existing, existing.ETag, TableUpdateMode.Replace, ct); + return; + } + catch (RequestFailedException ex) when (ex.Status == 412) + { + continue; + } + } + + throw new InvalidOperationException( + $"SoftwareInventory decrement ETag-CAS exhausted after {InventoryCasMaxAttempts} attempts or " + + $"{InventoryCasMaxWallClock.TotalSeconds:0}s wall-clock. tenant={tenantId} key={rowKey}"); + } + + // ============================================================ Helpers ==== + + /// + /// SoftwareInventory RowKey format — preserves the same shape used by the legacy + /// UpsertSoftwareInventoryAsync path so existing rows remain reachable by the + /// new helpers. SanitizeTableKey lives in TableStorageService.Rules.cs as a + /// private static; partial classes share scope. + /// + private static string BuildSoftwareInventoryRowKey(string vendor, string name, string version) + { + var raw = $"{vendor ?? string.Empty}:{name ?? string.Empty}:{version ?? string.Empty}"; + var sanitized = SanitizeTableKey(raw); + return sanitized.Length > 512 ? sanitized.Substring(0, 512) : sanitized; + } + } +} diff --git a/src/Backend/AutopilotMonitor.Functions/Services/TableStorageService.cs b/src/Backend/AutopilotMonitor.Functions/Services/TableStorageService.cs index 1ee82261..50eb2ae1 100644 --- a/src/Backend/AutopilotMonitor.Functions/Services/TableStorageService.cs +++ b/src/Backend/AutopilotMonitor.Functions/Services/TableStorageService.cs @@ -56,6 +56,17 @@ namespace AutopilotMonitor.Functions.Services } } + /// + /// Test seam: construct directly from a (possibly Moq'd) . + /// Used by xUnit so the storage-touching helpers in the partial classes (Deletion, + /// Inventory, …) can be exercised against the SDK's virtual surface without hitting Azure. + /// + internal TableStorageService(TableServiceClient tableServiceClient, ILogger logger) + { + _tableServiceClient = tableServiceClient; + _logger = logger; + } + /// /// Returns a TableClient for the specified table name. /// Used by services that need direct table access (e.g. VulnerabilityCorrelationService). diff --git a/src/Backend/AutopilotMonitor.Functions/Services/Vulnerability/VulnerabilityCorrelationService.cs b/src/Backend/AutopilotMonitor.Functions/Services/Vulnerability/VulnerabilityCorrelationService.cs index 1276c103..51e56bbc 100644 --- a/src/Backend/AutopilotMonitor.Functions/Services/Vulnerability/VulnerabilityCorrelationService.cs +++ b/src/Backend/AutopilotMonitor.Functions/Services/Vulnerability/VulnerabilityCorrelationService.cs @@ -7,6 +7,7 @@ using System.Threading; using System.Threading.Tasks; using AutopilotMonitor.Functions.DataAccess.TableStorage; using AutopilotMonitor.Shared.DataAccess; +using AutopilotMonitor.Shared.Models.Deletion; using Microsoft.ApplicationInsights; using Microsoft.Extensions.Logging; @@ -16,6 +17,7 @@ namespace AutopilotMonitor.Functions.Services.Vulnerability { private readonly IVulnerabilityRepository _vulnRepo; private readonly ISessionRepository _sessionRepo; + private readonly IMaintenanceRepository? _maintenanceRepo; private readonly NvdApiClient _nvdApi; private readonly KevDataService _kevService; private readonly MsrcCorrelationService _msrcCorrelation; @@ -42,10 +44,12 @@ namespace AutopilotMonitor.Functions.Services.Vulnerability MsrcCorrelationService msrcCorrelation, AdminConfigurationService adminConfig, TelemetryClient? telemetryClient, - ILogger logger) + ILogger logger, + IMaintenanceRepository? maintenanceRepo = null) { _vulnRepo = vulnRepo; _sessionRepo = sessionRepo; + _maintenanceRepo = maintenanceRepo; _nvdApi = nvdApi; _kevService = kevService; _msrcCorrelation = msrcCorrelation; @@ -242,18 +246,13 @@ namespace AutopilotMonitor.Functions.Services.Vulnerability return ra.CompareTo(rb); }); - // Persist software inventory (fire-and-forget, non-blocking) - _ = Task.Run(async () => - { - try - { - await _vulnRepo.UpsertSoftwareInventoryAsync(tenantId, inventoryItems, sessionId, cpeMappingsDict); - } - catch (Exception ex) - { - _logger.LogDebug(ex, "Failed to upsert software inventory (non-fatal)"); - } - }); + // Persist software inventory via the contributions side-row (Plan §17). + // Fire-and-forget so the report still gets stored if the side-row write hits a + // transient storage failure — the report is the primary product, inventory is the + // secondary tenant-aggregate signal. PR3 wires SessionDeletionGuard above this + // layer (at VulnerabilityCorrelateHandler) so a session locked by a cascade never + // reaches this code path; PR2's side-row write proceeds unguarded. + _ = Task.Run(() => ApplyInventoryContributionsAsync(tenantId, sessionId, inventoryItems, cpeMappingsDict)); // Summary log for production monitoring _logger.LogInformation( @@ -957,6 +956,162 @@ namespace AutopilotMonitor.Functions.Services.Vulnerability return results; } + /// + /// Side-row + counter write path (Plan §17). Reads the previous contributions row, + /// computes the (added, removed) delta against the new inventory, then applies + /// +1 per added (with full metadata for first-insert) and -1 per removed + /// (clamp ≥ 0). + /// + /// Per Plan §17.3 "Side-row write before counter writes" + §17.8: individual counter + /// failures must NOT abort the rest of the loop, because the side-row already promises + /// the new key set — if we stopped on first failure, the next correlation run would see + /// an empty delta and never retry the under-counted keys. Each counter call is therefore + /// guarded individually; failures emit an inventory_counter_write_failed audit + /// entry (when the maintenance repo is wired) and continue processing. The wrapping + /// catch handles only the side-row CAS exhaustion + the parser path. + /// + /// + internal async Task ApplyInventoryContributionsAsync( + string tenantId, string sessionId, + List> inventoryItems, + Dictionary? cpeMappings) + { + InventoryContributionsDelta delta; + try + { + var typedItems = ParseSoftwareInventoryItems(inventoryItems, cpeMappings); + if (typedItems.Count == 0) return; + delta = await _vulnRepo.UpsertSessionInventoryContributionsAsync(tenantId, sessionId, typedItems); + } + catch (Exception ex) + { + _logger.LogWarning(ex, + "Failed to upsert software inventory contributions side-row (non-fatal). tenant={TenantId} session={SessionId}", + tenantId, sessionId); + return; + } + + foreach (var added in delta.AddedItems) + { + try + { + await _vulnRepo.IncrementSoftwareInventoryEntryAsync(tenantId, added, sessionId); + } + catch (Exception ex) + { + _logger.LogWarning(ex, + "SoftwareInventory increment failed (non-fatal — counter drift accepted). tenant={TenantId} session={SessionId} vendor={Vendor} name={Name} version={Version}", + tenantId, sessionId, added.NormalizedVendor, added.NormalizedName, added.NormalizedVersion); + await TryAuditInventoryCounterFailureAsync( + tenantId, sessionId, "increment", + added.NormalizedVendor, added.NormalizedName, added.NormalizedVersion, ex); + } + } + + foreach (var removed in delta.RemovedKeys) + { + try + { + await _vulnRepo.DecrementSoftwareInventoryEntryAsync( + tenantId, removed.Vendor, removed.Name, removed.Version); + } + catch (Exception ex) + { + _logger.LogWarning(ex, + "SoftwareInventory decrement failed (non-fatal — counter drift accepted). tenant={TenantId} session={SessionId} vendor={Vendor} name={Name} version={Version}", + tenantId, sessionId, removed.Vendor, removed.Name, removed.Version); + await TryAuditInventoryCounterFailureAsync( + tenantId, sessionId, "decrement", + removed.Vendor, removed.Name, removed.Version, ex); + } + } + } + + /// + /// Emits the inventory_counter_write_failed audit entry on a single counter + /// failure (Plan §17.8). Non-fatal — audit emission itself must not break the loop. + /// + private async Task TryAuditInventoryCounterFailureAsync( + string tenantId, string sessionId, string operation, + string vendor, string name, string version, Exception failureCause) + { + if (_maintenanceRepo == null) return; + try + { + await _maintenanceRepo.LogAuditEntryAsync( + tenantId, + action: "inventory_counter_write_failed", + entityType: "Session", + entityId: sessionId, + performedBy: "system", + details: new Dictionary(StringComparer.Ordinal) + { + ["operation"] = operation, + ["vendor"] = vendor, + ["name"] = name, + ["version"] = version, + ["exceptionType"] = failureCause.GetType().Name, + ["exceptionMessage"] = failureCause.Message, + }); + } + catch (Exception ex) + { + _logger.LogDebug(ex, + "Failed to emit inventory_counter_write_failed audit entry (suppressed). tenant={TenantId} session={SessionId}", + tenantId, sessionId); + } + } + + /// + /// Materializes typed entries from the loose + /// inventoryItems dictionary shape carried over the wire. Skips entries with no + /// usable normalization triple (matches the existing dedup behaviour in + /// UpsertSoftwareInventoryAsync). Attaches the CPE URI from + /// when one is available for the normalized name — the increment helper then stamps it + /// on the SoftwareInventory row at first-insert. + /// + internal static List ParseSoftwareInventoryItems( + List> inventoryItems, + Dictionary? cpeMappings) + { + var items = new List(inventoryItems.Count); + foreach (var raw in inventoryItems) + { + var normalizedName = raw.ContainsKey("normalizedName") ? raw["normalizedName"]?.ToString() ?? "" : ""; + var normalizedVendor = raw.ContainsKey("normalizedPublisher") ? raw["normalizedPublisher"]?.ToString() ?? "" : ""; + var normalizedVersion = raw.ContainsKey("normalizedVersion") ? raw["normalizedVersion"]?.ToString() ?? "" : ""; + + if (string.IsNullOrEmpty(normalizedVendor) + && string.IsNullOrEmpty(normalizedName) + && string.IsNullOrEmpty(normalizedVersion)) + { + continue; + } + + string? cpeUri = null; + if (cpeMappings != null + && !string.IsNullOrEmpty(normalizedName) + && cpeMappings.TryGetValue(normalizedName, out var mapped) + && !string.IsNullOrEmpty(mapped)) + { + cpeUri = mapped; + } + + items.Add(new SoftwareInventoryItem + { + NormalizedVendor = normalizedVendor, + NormalizedName = normalizedName, + NormalizedVersion = normalizedVersion, + DisplayName = raw.ContainsKey("displayName") ? raw["displayName"]?.ToString() ?? "" : "", + Publisher = raw.ContainsKey("publisher") ? raw["publisher"]?.ToString() ?? "" : "", + RegistrySource = raw.ContainsKey("registrySource") ? raw["registrySource"]?.ToString() ?? "" : "", + NormalizationConfidence = raw.ContainsKey("normalizationConfidence") ? raw["normalizationConfidence"]?.ToString() ?? "" : "", + CpeUri = cpeUri, + }); + } + return items; + } + /// /// Replace characters not allowed in Azure Table Storage keys (/, \, #, ?). /// diff --git a/src/Shared/AutopilotMonitor.Shared/DataAccess/IVulnerabilityRepository.cs b/src/Shared/AutopilotMonitor.Shared/DataAccess/IVulnerabilityRepository.cs index 869de008..06550db1 100644 --- a/src/Shared/AutopilotMonitor.Shared/DataAccess/IVulnerabilityRepository.cs +++ b/src/Shared/AutopilotMonitor.Shared/DataAccess/IVulnerabilityRepository.cs @@ -1,5 +1,7 @@ using System.Collections.Generic; +using System.Threading; using System.Threading.Tasks; +using AutopilotMonitor.Shared.Models.Deletion; namespace AutopilotMonitor.Shared.DataAccess { @@ -27,6 +29,48 @@ namespace AutopilotMonitor.Shared.DataAccess Task>> GetAllSoftwareInventoryAsync(); Task UpdateSoftwareInventoryCpeAsync(string normalizedName, string cpeUri); + // --- Software Inventory v2: contributions side-row + delta-driven counters (Plan §17) --- + + /// + /// Reads the SessionInventoryContributions side-row for a session. Returns the + /// previously-counted normalized keys, or null when no side-row exists yet + /// (pre-side-row session OR first inventory correlation for this session). + /// + Task?> GetSessionInventoryContributionsAsync( + string tenantId, string sessionId, CancellationToken ct = default); + + /// + /// At-most-once-per-session, delta-update upsert of the side-row. Plan §17.3 ETag-CAS + /// loop: reads side-row, computes (added, removed), writes the new key set back with + /// ETag-Replace (or AddEntity on first call). Returns the delta the caller must apply + /// to the aggregate counter. Fail-loud on bounded-retry exhaustion (10 attempts or 60s + /// wall-clock per §12-Q10). + /// + Task UpsertSessionInventoryContributionsAsync( + string tenantId, string sessionId, + IReadOnlyList newItems, + CancellationToken ct = default); + + /// Deletes the side-row (cascade tombstone step). 404 = treated as already gone. + Task DeleteSessionInventoryContributionsAsync(string tenantId, string sessionId, CancellationToken ct = default); + + /// + /// ETag-CAS SessionCount += 1 on the SoftwareInventory row for the given + /// normalized key. Creates the row with full metadata on first encounter. Bounded retry + /// (10 attempts / 60s wall-clock); throws on exhaustion. + /// + Task IncrementSoftwareInventoryEntryAsync( + string tenantId, SoftwareInventoryItem item, string sessionId, CancellationToken ct = default); + + /// + /// ETag-CAS SessionCount -= 1 with clamp-≥-0 on the matching SoftwareInventory + /// row. Deletes the row if the count hits zero. 404 = no-op (idempotent). Bounded retry + /// (10 attempts / 60s); throws on exhaustion. + /// + Task DecrementSoftwareInventoryEntryAsync( + string tenantId, string normalizedVendor, string normalizedName, string normalizedVersion, + CancellationToken ct = default); + // --- CPE Mappings --- Task ImportCpeMappingSeedFromJsonAsync(string json); Task ImportCpeCommunityMappingsFromJsonAsync(string json); diff --git a/src/Shared/AutopilotMonitor.Shared/Models/Deletion/SoftwareInventoryContributions.cs b/src/Shared/AutopilotMonitor.Shared/Models/Deletion/SoftwareInventoryContributions.cs new file mode 100644 index 00000000..310e35bf --- /dev/null +++ b/src/Shared/AutopilotMonitor.Shared/Models/Deletion/SoftwareInventoryContributions.cs @@ -0,0 +1,188 @@ +using System; +using System.Collections.Generic; +using System.IO; +using System.IO.Compression; +using System.Linq; +using System.Text; +using System.Text.Json; + +namespace AutopilotMonitor.Shared.Models.Deletion +{ + /// + /// Single software entry as it arrives from a software_inventory_analysis event. + /// PR2: the unit that identifies for cascade-time + /// decrement, plus the metadata needed to (re-)create a fresh SoftwareInventory row + /// when this is the first session that's seen the software. + /// + public class SoftwareInventoryItem + { + // Normalized fields used as the identity key. The triple (Vendor, Name, Version) is + // the side-row key and the SoftwareInventory RowKey component. + public string NormalizedVendor { get; set; } = string.Empty; + public string NormalizedName { get; set; } = string.Empty; + public string NormalizedVersion { get; set; } = string.Empty; + + // Display metadata — only written on first-insert into SoftwareInventory; later + // correlations only bump SessionCount + LastSeenAt + LastSessionId. + public string DisplayName { get; set; } = string.Empty; + public string Publisher { get; set; } = string.Empty; + public string RegistrySource { get; set; } = string.Empty; + public string NormalizationConfidence { get; set; } = string.Empty; + + /// + /// Optional CPE URI for this software (resolved by CpeMappingService at correlation + /// time). Carried so the increment helper can stamp it on the row at first-insert. + /// + public string? CpeUri { get; set; } + + /// The canonical key used in the contributions side-row JSON and as the SoftwareInventory RowKey component. + public DeletionDecrementKey ToDecrementKey() => new DeletionDecrementKey + { + Vendor = NormalizedVendor, + Name = NormalizedName, + Version = NormalizedVersion, + }; + } + + /// + /// Result of a UpsertSessionInventoryContributionsAsync call: the items added vs the + /// keys removed compared to the previous side-row state. Caller applies +1 per added + /// (with full metadata for first-insert) and -1 per removed (key only, decrement + /// clamps at zero). + /// + public class InventoryContributionsDelta + { + public List AddedItems { get; set; } = new List(); + public List RemovedKeys { get; set; } = new List(); + + /// True when this was the first time the side-row was created for the session. + public bool FirstTime { get; set; } + } + + /// + /// JSON + gzip+Base64 codec for the SoftwareKeysJson column on the + /// SessionInventoryContributions side-row. Pure functions — easy to unit-test + /// independent of any Azure SDK plumbing. + /// + /// Plan §17.7: raw JSON for small key sets, gzip+Base64 once the raw payload exceeds + /// . IsCompressed on the row chooses the + /// decode path. uses this same codec for the + /// cascade-time decrement-key extraction (PR1 already wired). + /// + /// + public static class SoftwareKeysJsonCodec + { + public const int CompressionThresholdBytes = 30_000; + + public static EncodedSoftwareKeys Encode(IEnumerable keys) + { + var ordered = keys + .OrderBy(k => k.Vendor, StringComparer.Ordinal) + .ThenBy(k => k.Name, StringComparer.Ordinal) + .ThenBy(k => k.Version, StringComparer.Ordinal) + .ToList(); + + var raw = JsonSerializer.Serialize(ordered, DeletionManifestJson.SerializerOptions); + if (raw.Length <= CompressionThresholdBytes) + { + return new EncodedSoftwareKeys { Encoded = raw, IsCompressed = false, KeyCount = ordered.Count }; + } + + using var output = new MemoryStream(); + using (var gzip = new GZipStream(output, CompressionLevel.Optimal, leaveOpen: true)) + { + var bytes = Encoding.UTF8.GetBytes(raw); + gzip.Write(bytes, 0, bytes.Length); + } + var compressed = Convert.ToBase64String(output.ToArray()); + return new EncodedSoftwareKeys { Encoded = compressed, IsCompressed = true, KeyCount = ordered.Count }; + } + + public static List Decode(string encoded, bool isCompressed) + { + if (string.IsNullOrEmpty(encoded)) return new List(); + + string json; + if (isCompressed) + { + byte[] gz; + try + { + gz = Convert.FromBase64String(encoded); + } + catch (FormatException ex) + { + throw new InvalidDataException( + "SoftwareKeysJson is marked IsCompressed=true but the payload is not valid Base64.", ex); + } + using var input = new MemoryStream(gz); + using var gunzip = new GZipStream(input, CompressionMode.Decompress); + using var output = new MemoryStream(); + gunzip.CopyTo(output); + json = Encoding.UTF8.GetString(output.ToArray()); + } + else + { + json = encoded; + } + + try + { + return JsonSerializer.Deserialize>(json, DeletionManifestJson.SerializerOptions) + ?? new List(); + } + catch (JsonException ex) + { + throw new InvalidDataException( + "SoftwareKeysJson is not valid JSON for List.", ex); + } + } + + /// + /// Compares against and returns the + /// (added, removed) delta. Comparison is case-insensitive on every component because + /// upstream normalization is best-effort across registry sources. + /// + public static (HashSet Added, HashSet Removed) ComputeDelta( + IEnumerable oldKeys, + IEnumerable newKeys) + { + var oldSet = ToCompositeSet(oldKeys); + var newSet = ToCompositeSet(newKeys); + var added = new HashSet(newSet, StringComparer.OrdinalIgnoreCase); + added.ExceptWith(oldSet); + var removed = new HashSet(oldSet, StringComparer.OrdinalIgnoreCase); + removed.ExceptWith(newSet); + return (added, removed); + } + + /// + /// Composite-key string used as a lookup identity. Three-part null-safe join with a + /// separator character that cannot legitimately appear in any normalized component + /// (we use '', START OF HEADING — never present in publisher / name / version). + /// + public static string CompositeKey(string vendor, string name, string version) + => $"{vendor}{name}{version}"; + + public static string CompositeKey(DeletionDecrementKey key) + => CompositeKey(key.Vendor ?? string.Empty, key.Name ?? string.Empty, key.Version ?? string.Empty); + + private static HashSet ToCompositeSet(IEnumerable keys) + { + var set = new HashSet(StringComparer.OrdinalIgnoreCase); + foreach (var k in keys) + { + set.Add(CompositeKey(k)); + } + return set; + } + } + + /// Encoded form of a software-keys JSON payload, tagged with the codec flag. + public class EncodedSoftwareKeys + { + public string Encoded { get; set; } = string.Empty; + public bool IsCompressed { get; set; } + public int KeyCount { get; set; } + } +}