From ee07059bc370618f88bd328fcfcd5ea9ef7d2c33 Mon Sep 17 00:00:00 2001 From: Oliver Kieselbach Date: Tue, 21 Apr 2026 14:13:07 +0200 Subject: [PATCH] fix(backend): harden /api/agent/telemetry batch validation (Codex Finding 3) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The ingest endpoint trusted the first item's PartitionKey for the whole batch and silently defaulted RowKey-driver fields to 0 when the agent envelope omitted them. Both paths can misroute data on the storage layer. 1. Cross-session guard: every TelemetryItemDto in a batch must share the same PartitionKey as items[0]. Without this, a mixed-session batch would be stamped with session-A's identity inside PersistItemsAsync and silently relocate session-B's signals/transitions into session-A's primary rows. Mismatch now returns 400 BadRequest with the offending index logged. Extracted as pure helper FindMismatchingPartitionKey so the guard is unit-testable without an HTTP harness. 2. RowKey-driver enforcement in TelemetryPayloadParser: - ParseSignal now rejects envelopes where SessionSignalOrdinal is absent or null (returns null → caller drops the item). A missing value would default to 0 and clobber a legitimate ordinal-0 row on UpsertReplace. Explicit ordinal=0 stays a valid value. - ParseTransition applies the same defense to StepIndex. - Distinguishes "key missing" from "key present with value 0" via JObject.TryGetValue + JTokenType check. Tests (+11 in Functions): - IngestTelemetryFunctionTests: empty batch, single-item batch, uniform batch, mismatch in middle, case-sensitive mismatch. - TelemetryPayloadParserTests: missing/null SessionSignalOrdinal on Signal → null; explicit ordinal 0 accepted; missing/null StepIndex on DecisionTransition → null; explicit StepIndex 0 accepted. Full Functions suite: 703 green (prev 692 + 11 new). Does not address Codex Findings 1 (V2 agent has no Signal/Transition upload path) or 2 (AdminAction=Succeeded mis-mapped to Failed locally) — those land in follow-up commits on this branch. Co-Authored-By: Claude Opus 4.7 (1M context) --- .../IngestTelemetryFunctionTests.cs | 80 +++++++++++++++++++ .../TelemetryPayloadParserTests.cs | 51 ++++++++++++ .../Ingest/IngestTelemetryFunction.cs | 46 ++++++++++- .../Ingest/TelemetryPayloadParser.cs | 55 ++++++++++++- 4 files changed, 229 insertions(+), 3 deletions(-) diff --git a/src/Backend/AutopilotMonitor.Functions.Tests/IngestTelemetryFunctionTests.cs b/src/Backend/AutopilotMonitor.Functions.Tests/IngestTelemetryFunctionTests.cs index 5d8455a3..15ff904f 100644 --- a/src/Backend/AutopilotMonitor.Functions.Tests/IngestTelemetryFunctionTests.cs +++ b/src/Backend/AutopilotMonitor.Functions.Tests/IngestTelemetryFunctionTests.cs @@ -1,6 +1,7 @@ using System.IO; using System.Text; using AutopilotMonitor.Functions.Functions.Ingest; +using AutopilotMonitor.Shared.Models; namespace AutopilotMonitor.Functions.Tests; @@ -95,4 +96,83 @@ public class IngestTelemetryFunctionTests Assert.False(exceeded); Assert.Equal(string.Empty, body); } + + // ============================================================ Batch PartitionKey uniformity + + private static TelemetryItemDto Item(string partitionKey) + => new TelemetryItemDto + { + Kind = "Signal", + PartitionKey = partitionKey, + RowKey = "0000000001", + TelemetryItemId = 1, + PayloadJson = "{}", + EnqueuedAtUtc = DateTime.UtcNow, + }; + + [Fact] + public void FindMismatchingPartitionKey_returns_false_for_empty_batch() + { + var result = IngestTelemetryFunction.FindMismatchingPartitionKey( + new List(), out var idx, out var value); + + Assert.False(result); + Assert.Equal(-1, idx); + Assert.Null(value); + } + + [Fact] + public void FindMismatchingPartitionKey_returns_false_for_single_item_batch() + { + var result = IngestTelemetryFunction.FindMismatchingPartitionKey( + new[] { Item($"{TenantId}_{SessionId}") }, out var idx, out var value); + + Assert.False(result); + Assert.Equal(-1, idx); + Assert.Null(value); + } + + [Fact] + public void FindMismatchingPartitionKey_returns_false_when_all_items_share_PartitionKey() + { + var pk = $"{TenantId}_{SessionId}"; + var items = new[] { Item(pk), Item(pk), Item(pk) }; + + var result = IngestTelemetryFunction.FindMismatchingPartitionKey(items, out var idx, out var value); + + Assert.False(result); + Assert.Equal(-1, idx); + Assert.Null(value); + } + + [Fact] + public void FindMismatchingPartitionKey_returns_index_and_value_on_mismatch() + { + // Session-mismatch in the middle of the batch — what a misbehaving client could send. + var first = $"{TenantId}_{SessionId}"; + var second = $"{TenantId}_c3d4e5f6-a7b8-9012-cdef-123456789012"; // same tenant, different session + var items = new[] { Item(first), Item(first), Item(second) }; + + var result = IngestTelemetryFunction.FindMismatchingPartitionKey(items, out var idx, out var value); + + Assert.True(result); + Assert.Equal(2, idx); + Assert.Equal(second, value); + } + + [Fact] + public void FindMismatchingPartitionKey_is_case_sensitive() + { + // PartitionKeys are GUIDs — matching is ordinal. A lowercase/uppercase-swapped second + // item is treated as a different partition (safe-by-default; the agent emits lowercase). + var first = $"{TenantId}_{SessionId}"; + var second = first.ToUpperInvariant(); + var items = new[] { Item(first), Item(second) }; + + var result = IngestTelemetryFunction.FindMismatchingPartitionKey(items, out var idx, out var value); + + Assert.True(result); + Assert.Equal(1, idx); + Assert.Equal(second, value); + } } diff --git a/src/Backend/AutopilotMonitor.Functions.Tests/TelemetryPayloadParserTests.cs b/src/Backend/AutopilotMonitor.Functions.Tests/TelemetryPayloadParserTests.cs index de7e1eb0..f9465881 100644 --- a/src/Backend/AutopilotMonitor.Functions.Tests/TelemetryPayloadParserTests.cs +++ b/src/Backend/AutopilotMonitor.Functions.Tests/TelemetryPayloadParserTests.cs @@ -126,6 +126,30 @@ public class TelemetryPayloadParserTests Assert.Equal(99L, rec!.SessionTraceOrdinal); } + [Theory] + [InlineData("{\"Kind\":\"SessionStarted\"}")] // key absent + [InlineData("{\"SessionSignalOrdinal\":null,\"Kind\":\"SessionStarted\"}")] // key present but null + public void ParseSignal_drops_item_when_SessionSignalOrdinal_is_missing_or_null(string payload) + { + // A missing ordinal would silently default to 0 → collide with a legitimate ordinal-0 + // row on UpsertReplace. The parser must reject it so the caller loop drops the item. + var rec = TelemetryPayloadParser.ParseSignal(Dto("Signal", payload), TenantId, SessionId); + Assert.Null(rec); + } + + [Fact] + public void ParseSignal_accepts_explicit_ordinal_zero() + { + // 0 is a legal value (the agent seeds ordinals from 0 on a fresh SignalLog). Only + // absent/null is rejected — an explicit 0 must round-trip. + var payload = "{\"SessionSignalOrdinal\":0,\"Kind\":\"SessionStarted\"}"; + + var rec = TelemetryPayloadParser.ParseSignal(Dto("Signal", payload), TenantId, SessionId); + + Assert.NotNull(rec); + Assert.Equal(0L, rec!.SessionSignalOrdinal); + } + // ============================================================ DecisionTransition [Fact] @@ -173,6 +197,33 @@ public class TelemetryPayloadParserTests } } + [Theory] + [InlineData("{\"SessionTraceOrdinal\":1,\"ToStage\":\"EspInProgress\"}")] // key absent + [InlineData("{\"StepIndex\":null,\"SessionTraceOrdinal\":1,\"ToStage\":\"EspInProgress\"}")] // key present but null + public void ParseTransition_drops_item_when_StepIndex_is_missing_or_null(string payload) + { + // Same defense as ParseSignal: a missing StepIndex would default to 0 → collide with the + // legitimate step-0 row on UpsertReplace. + var rec = TelemetryPayloadParser.ParseTransition(Dto("DecisionTransition", payload), TenantId, SessionId); + Assert.Null(rec); + } + + [Fact] + public void ParseTransition_accepts_explicit_stepIndex_zero() + { + // Step indices start at 0 (the first reducer call produces StepIndex=0). An explicit 0 + // must not be confused with a missing key. + var payload = + "{\"StepIndex\":0,\"SessionTraceOrdinal\":1,\"SignalOrdinalRef\":1," + + "\"Trigger\":\"t\",\"FromStage\":\"Pending\",\"ToStage\":\"EspInProgress\",\"Taken\":true," + + "\"ReducerVersion\":\"1.0.0\",\"OccurredAtUtc\":\"2026-04-21T10:00:00Z\"}"; + + var rec = TelemetryPayloadParser.ParseTransition(Dto("DecisionTransition", payload), TenantId, SessionId); + + Assert.NotNull(rec); + Assert.Equal(0, rec!.StepIndex); + } + [Fact] public void ParseTransition_extracts_classifier_verdict_nested_fields() { diff --git a/src/Backend/AutopilotMonitor.Functions/Functions/Ingest/IngestTelemetryFunction.cs b/src/Backend/AutopilotMonitor.Functions/Functions/Ingest/IngestTelemetryFunction.cs index adf44e29..1f7d962d 100644 --- a/src/Backend/AutopilotMonitor.Functions/Functions/Ingest/IngestTelemetryFunction.cs +++ b/src/Backend/AutopilotMonitor.Functions/Functions/Ingest/IngestTelemetryFunction.cs @@ -160,7 +160,9 @@ namespace AutopilotMonitor.Functions.Functions.Ingest } // Extract tenant+session from the first item's PartitionKey for body-vs-header tenant check - // and for AdminAction / ServerAction lookups. All items in a batch belong to one session. + // and for AdminAction / ServerAction lookups. All items in a batch must belong to one + // session — anything else is a client bug and gets the whole batch rejected (see + // the uniformity check below). if (!TryParsePartitionKey(items[0].PartitionKey, out var bodyTenantId, out var sessionId)) { return AsOutput(await WriteErrorAsync(req, HttpStatusCode.BadRequest, "Malformed PartitionKey")); @@ -174,6 +176,20 @@ namespace AutopilotMonitor.Functions.Functions.Ingest return AsOutput(await WriteErrorAsync(req, HttpStatusCode.Forbidden, "TenantId mismatch between header and payload")); } + // Defense-in-depth: every item must carry the same PartitionKey as the first. Without + // this check a mixed-session batch would stamp every item with session A's identity + // inside PersistItemsAsync, silently relocating session B's signals/transitions into + // session A's primary rows. + if (FindMismatchingPartitionKey(items, out var mismatchIndex, out var mismatchedValue)) + { + _logger.LogWarning( + "IngestTelemetry: PartitionKey mismatch at item[{Index}] — expected={Expected}, got={Got}", + mismatchIndex, items[0].PartitionKey, mismatchedValue); + return AsOutput(await WriteErrorAsync( + req, HttpStatusCode.BadRequest, + "All telemetry items in a batch must share the same PartitionKey")); + } + // Partition + persist. Events are routed through EventIngestProcessor which runs the // full pipeline (rule engine / app-install aggregation / SignalR / webhooks / ...); // Signal + Transition go straight to their repositories. @@ -438,6 +454,34 @@ namespace AutopilotMonitor.Functions.Functions.Ingest return (false, System.Text.Encoding.UTF8.GetString(buffered.ToArray())); } + /// + /// Scans the batch for any item whose differs + /// from the first item's. Returns true + the offending index/value on mismatch; false if + /// every item shares the same PartitionKey (or the batch has < 2 items). Pure so the + /// guard can be unit-tested without a live HTTP trigger. + /// + internal static bool FindMismatchingPartitionKey( + IReadOnlyList items, + out int mismatchIndex, + out string? mismatchedValue) + { + mismatchIndex = -1; + mismatchedValue = null; + if (items is null || items.Count < 2) return false; + + var expected = items[0].PartitionKey; + for (var i = 1; i < items.Count; i++) + { + if (!string.Equals(items[i].PartitionKey, expected, StringComparison.Ordinal)) + { + mismatchIndex = i; + mismatchedValue = items[i].PartitionKey; + return true; + } + } + return false; + } + /// /// PartitionKey convention is {tenantId}_{sessionId} — both GUIDs with dashes, so /// splitting on the single underscore between them is unambiguous. Returns false for any diff --git a/src/Backend/AutopilotMonitor.Functions/Functions/Ingest/TelemetryPayloadParser.cs b/src/Backend/AutopilotMonitor.Functions/Functions/Ingest/TelemetryPayloadParser.cs index 9329ea69..2c6183ea 100644 --- a/src/Backend/AutopilotMonitor.Functions/Functions/Ingest/TelemetryPayloadParser.cs +++ b/src/Backend/AutopilotMonitor.Functions/Functions/Ingest/TelemetryPayloadParser.cs @@ -1,3 +1,4 @@ +using System; using AutopilotMonitor.Shared.Models; using Newtonsoft.Json; using Newtonsoft.Json.Linq; @@ -46,7 +47,11 @@ namespace AutopilotMonitor.Functions.Functions.Ingest try { root = JObject.Parse(item.PayloadJson!); } catch (JsonException) { return null; } - var ordinal = (long?)root["SessionSignalOrdinal"] ?? 0L; + // SessionSignalOrdinal is the RowKey driver — a missing value would silently default + // to 0 and clobber a legitimate ordinal-0 row on UpsertReplace. Require explicit + // presence; 0 itself is a valid value the caller may intentionally send. + if (!TryGetLong(root, "SessionSignalOrdinal", out var ordinal)) return null; + var traceOrdinal = (long?)root["SessionTraceOrdinal"] ?? item!.SessionTraceOrdinal ?? 0L; var kind = (string?)root["Kind"] ?? string.Empty; var schemaVersion = (int?)root["KindSchemaVersion"] ?? 0; @@ -80,7 +85,10 @@ namespace AutopilotMonitor.Functions.Functions.Ingest try { root = JObject.Parse(item.PayloadJson!); } catch (JsonException) { return null; } - var stepIndex = (int?)root["StepIndex"] ?? 0; + // StepIndex is the RowKey driver — same defense as SessionSignalOrdinal above. + // A missing StepIndex must not silently default to 0 and clobber the real step-0 row. + if (!TryGetInt(root, "StepIndex", out var stepIndex)) return null; + var traceOrdinal = (long?)root["SessionTraceOrdinal"] ?? item!.SessionTraceOrdinal ?? 0L; var signalOrdinal = (long?)root["SignalOrdinalRef"] ?? 0L; var occurredAt = (DateTime?)root["OccurredAtUtc"] ?? item!.EnqueuedAtUtc; @@ -142,5 +150,48 @@ namespace AutopilotMonitor.Functions.Functions.Ingest return false; } } + + /// + /// Reads a property that MUST be present on the envelope. Distinguishes "missing key" + /// (returns false) from "present with value 0" (returns true, value=0). Used for + /// RowKey-driving fields where a silent default would cause row collisions. + /// + private static bool TryGetLong(JObject root, string key, out long value) + { + if (root.TryGetValue(key, StringComparison.Ordinal, out var token) + && token.Type != JTokenType.Null + && token.Type != JTokenType.Undefined) + { + try + { + value = token.Value(); + return true; + } + catch (System.FormatException) { } + catch (System.OverflowException) { } + catch (System.InvalidCastException) { } + } + value = 0; + return false; + } + + private static bool TryGetInt(JObject root, string key, out int value) + { + if (root.TryGetValue(key, StringComparison.Ordinal, out var token) + && token.Type != JTokenType.Null + && token.Type != JTokenType.Undefined) + { + try + { + value = token.Value(); + return true; + } + catch (System.FormatException) { } + catch (System.OverflowException) { } + catch (System.InvalidCastException) { } + } + value = 0; + return false; + } } }