added unit test critical path (prevent future ingest errors)

This commit is contained in:
Oliver Kieselbach
2026-03-19 17:21:38 +01:00
parent 392892a501
commit 96fb8e1f03
9 changed files with 685 additions and 103 deletions
@@ -1 +1 @@
511
512
@@ -0,0 +1,221 @@
using AutopilotMonitor.Functions.Functions.Ingest;
using AutopilotMonitor.Functions.Security;
using AutopilotMonitor.Shared.Models;
namespace AutopilotMonitor.Functions.Tests;
/// <summary>
/// Tests for the critical ingest processing path.
///
/// REGRESSION GUARD: On 2026-03-19 the ingest endpoint returned 500 for all devices because
/// individual events did not have TenantId/SessionId set before being passed to
/// StoreEventsBatchAsync, which validates them as GUIDs.
/// StampServerFields (extracted from ProcessIngestAsync) is the fix — these tests ensure
/// it can never be accidentally removed or bypassed.
/// </summary>
public class IngestCriticalPathTests
{
private static readonly string ValidTenantId = "a1b2c3d4-e5f6-7890-abcd-ef1234567890";
private static readonly string ValidSessionId = "b2c3d4e5-f6a7-8901-bcde-f12345678901";
// -------------------------------------------------------------------------
// StampServerFields — THE regression test
// -------------------------------------------------------------------------
[Fact]
public void StampServerFields_SetsReceivedAtOnAllEvents()
{
var receivedAt = DateTime.UtcNow;
var events = MakeEvents(3);
IngestEventsFunction.StampServerFields(events, ValidTenantId, ValidSessionId, receivedAt);
Assert.All(events, evt => Assert.Equal(receivedAt, evt.ReceivedAt));
}
[Fact]
public void StampServerFields_SetsTenantIdOnAllEvents()
{
var events = MakeEvents(3);
IngestEventsFunction.StampServerFields(events, ValidTenantId, ValidSessionId, DateTime.UtcNow);
Assert.All(events, evt => Assert.Equal(ValidTenantId, evt.TenantId));
}
[Fact]
public void StampServerFields_SetsSessionIdOnAllEvents()
{
var events = MakeEvents(3);
IngestEventsFunction.StampServerFields(events, ValidTenantId, ValidSessionId, DateTime.UtcNow);
Assert.All(events, evt => Assert.Equal(ValidSessionId, evt.SessionId));
}
[Fact]
public void StampServerFields_OverridesNullTenantId()
{
// Regression: agent may not set TenantId per-event (e.g. new event types, early startup)
var events = new List<EnrollmentEvent>
{
new() { TenantId = null, SessionId = null },
};
IngestEventsFunction.StampServerFields(events, ValidTenantId, ValidSessionId, DateTime.UtcNow);
Assert.Equal(ValidTenantId, events[0].TenantId);
Assert.Equal(ValidSessionId, events[0].SessionId);
}
[Fact]
public void StampServerFields_OverridesWrongTenantId()
{
// Regression: agent running in unusual context may send wrong or empty TenantId per-event
var events = new List<EnrollmentEvent>
{
new() { TenantId = "wrong-tenant-value", SessionId = "wrong-session" },
};
IngestEventsFunction.StampServerFields(events, ValidTenantId, ValidSessionId, DateTime.UtcNow);
Assert.Equal(ValidTenantId, events[0].TenantId);
Assert.Equal(ValidSessionId, events[0].SessionId);
}
[Fact]
public void StampServerFields_AllEventsReceiveSameTenantId()
{
// All events in one batch belong to the same tenant — never mixed
var events = new List<EnrollmentEvent>
{
new() { TenantId = null },
new() { TenantId = "different-1" },
new() { TenantId = "different-2" },
new() { TenantId = ValidTenantId }, // already correct
};
IngestEventsFunction.StampServerFields(events, ValidTenantId, ValidSessionId, DateTime.UtcNow);
Assert.All(events, evt => Assert.Equal(ValidTenantId, evt.TenantId));
}
[Fact]
public void StampServerFields_EmptyList_DoesNotThrow()
{
var ex = Record.Exception(() =>
IngestEventsFunction.StampServerFields([], ValidTenantId, ValidSessionId, DateTime.UtcNow));
Assert.Null(ex);
}
// -------------------------------------------------------------------------
// Guard: after StampServerFields, EnsureValidGuid must pass for all events
// (simulates the contract between ProcessIngestAsync and StoreEventsBatchAsync)
// -------------------------------------------------------------------------
[Fact]
public void StampServerFields_ThenEnsureValidGuid_DoesNotThrow()
{
// This is the exact sequence in ProcessIngestAsync → StoreEventsBatchAsync.
// If StampServerFields is removed or broken, this test fails with ArgumentException.
var events = MakeEvents(5, tenantId: null, sessionId: null);
IngestEventsFunction.StampServerFields(events, ValidTenantId, ValidSessionId, DateTime.UtcNow);
// Simulate what StoreEventsBatchAsync does:
foreach (var evt in events)
{
SecurityValidator.EnsureValidGuid(evt.TenantId, "TenantId");
SecurityValidator.EnsureValidGuid(evt.SessionId, "SessionId");
}
// No exception = pass
}
[Fact]
public void WithoutStampServerFields_EnsureValidGuid_Throws()
{
// Documents WHY StampServerFields is required: without it, storage validation explodes.
var events = MakeEvents(1, tenantId: null, sessionId: null);
Assert.Throws<ArgumentException>(() =>
SecurityValidator.EnsureValidGuid(events[0].TenantId, "TenantId"));
}
// -------------------------------------------------------------------------
// NDJSON → Stamp → Validate: full pipeline test (no I/O, pure logic)
// -------------------------------------------------------------------------
[Fact]
public void FullPipeline_ParseNdjsonThenStamp_AllEventsHaveValidGuids()
{
// Simulates what ProcessIngestAsync does from parse to storage:
// 1. Parse NDJSON (events may have null/wrong TenantId from agent)
// 2. StampServerFields (backend overrides with validated tenantId)
// 3. EnsureValidGuid passes → no 500
var ndjsonLines = new[]
{
Newtonsoft.Json.JsonConvert.SerializeObject(
new { SessionId = ValidSessionId, TenantId = ValidTenantId }),
Newtonsoft.Json.JsonConvert.SerializeObject(
new EnrollmentEvent { EventType = "phase_changed" }), // no TenantId
Newtonsoft.Json.JsonConvert.SerializeObject(
new EnrollmentEvent { EventType = "hardware_spec", TenantId = "" }), // empty TenantId
};
var ndjson = string.Join('\n', ndjsonLines);
var (sessionId, tenantId, events) = NdjsonParser.ParseNdjson(ndjson);
// Stamp (as ProcessIngestAsync does)
IngestEventsFunction.StampServerFields(events, tenantId, sessionId, DateTime.UtcNow);
// Validate (as StoreEventsBatchAsync does)
foreach (var evt in events)
{
SecurityValidator.EnsureValidGuid(evt.TenantId, "TenantId");
SecurityValidator.EnsureValidGuid(evt.SessionId, "SessionId");
}
// No exception = the regression is fixed
}
// -------------------------------------------------------------------------
// TenantId mismatch guard (body vs header)
// -------------------------------------------------------------------------
[Fact]
public void TenantIdMismatch_DifferentCasing_IsDetected()
{
// OrdinalIgnoreCase is used — same GUID in different case should NOT be a mismatch
var header = ValidTenantId.ToLowerInvariant();
var body = ValidTenantId.ToUpperInvariant();
Assert.True(string.Equals(header, body, StringComparison.OrdinalIgnoreCase),
"Same GUID in different casing must be treated as equal (prevents false 403)");
}
[Fact]
public void TenantIdMismatch_DifferentGuids_IsDetected()
{
var headerTenant = ValidTenantId;
var bodyTenant = "c3d4e5f6-a7b8-9012-cdef-012345678901";
Assert.False(string.Equals(headerTenant, bodyTenant, StringComparison.OrdinalIgnoreCase),
"Different GUIDs must be detected as a mismatch (prevents TenantId spoofing)");
}
// -------------------------------------------------------------------------
// Helpers
// -------------------------------------------------------------------------
private static List<EnrollmentEvent> MakeEvents(int count, string? tenantId = "original-tenant", string? sessionId = "original-session")
{
return Enumerable.Range(0, count)
.Select(i => new EnrollmentEvent
{
EventType = $"test_event_{i}",
TenantId = tenantId,
SessionId = sessionId
})
.ToList();
}
}
@@ -0,0 +1,190 @@
using AutopilotMonitor.Functions.Functions.Ingest;
using AutopilotMonitor.Shared.Models;
using Newtonsoft.Json;
namespace AutopilotMonitor.Functions.Tests;
/// <summary>
/// Tests for NdjsonParser — the NDJSON+gzip parsing contract.
/// If the wire format between agent and backend ever drifts, these tests will catch it.
/// </summary>
public class NdjsonParserTests
{
private static readonly string ValidTenantId = "a1b2c3d4-e5f6-7890-abcd-ef1234567890";
private static readonly string ValidSessionId = "b2c3d4e5-f6a7-8901-bcde-f12345678901";
// --- ParseNdjson (plain string, no gzip) ---
[Fact]
public void ParseNdjson_MetadataOnly_ReturnsCorrectTenantAndSession()
{
var ndjson = JsonConvert.SerializeObject(new { SessionId = ValidSessionId, TenantId = ValidTenantId });
var (sessionId, tenantId, events) = NdjsonParser.ParseNdjson(ndjson);
Assert.Equal(ValidSessionId, sessionId);
Assert.Equal(ValidTenantId, tenantId);
Assert.Empty(events);
}
[Fact]
public void ParseNdjson_WithEvents_ReturnsAllEvents()
{
var lines = new[]
{
JsonConvert.SerializeObject(new { SessionId = ValidSessionId, TenantId = ValidTenantId }),
JsonConvert.SerializeObject(new EnrollmentEvent { EventType = "phase_changed", Message = "Device Setup" }),
JsonConvert.SerializeObject(new EnrollmentEvent { EventType = "app_install_started", Message = "MyApp" }),
};
var ndjson = string.Join('\n', lines);
var (_, _, events) = NdjsonParser.ParseNdjson(ndjson);
Assert.Equal(2, events.Count);
Assert.Equal("phase_changed", events[0].EventType);
Assert.Equal("app_install_started", events[1].EventType);
}
[Fact]
public void ParseNdjson_EventsDoNotInheritTenantIdFromMetadata()
{
// Critical contract: the parser does NOT stamp TenantId onto events.
// That is done by StampServerFields — these are two separate responsibilities.
var lines = new[]
{
JsonConvert.SerializeObject(new { SessionId = ValidSessionId, TenantId = ValidTenantId }),
JsonConvert.SerializeObject(new EnrollmentEvent { EventType = "test_event" }),
};
var ndjson = string.Join('\n', lines);
var (_, _, events) = NdjsonParser.ParseNdjson(ndjson);
// Event has no TenantId from agent JSON → will be null until StampServerFields is called
Assert.Null(events[0].TenantId);
Assert.Null(events[0].SessionId);
}
[Fact]
public void ParseNdjson_EventWithAgentSuppliedTenantId_PreservesItUntilStamped()
{
var agentTenantId = "c3d4e5f6-a7b8-9012-cdef-012345678901";
var lines = new[]
{
JsonConvert.SerializeObject(new { SessionId = ValidSessionId, TenantId = ValidTenantId }),
JsonConvert.SerializeObject(new EnrollmentEvent { EventType = "test_event", TenantId = agentTenantId }),
};
var ndjson = string.Join('\n', lines);
var (_, _, events) = NdjsonParser.ParseNdjson(ndjson);
// Parser preserves what the agent sent — StampServerFields will override it
Assert.Equal(agentTenantId, events[0].TenantId);
}
[Fact]
public void ParseNdjson_TrailingNewlines_AreIgnored()
{
var ndjson = JsonConvert.SerializeObject(new { SessionId = ValidSessionId, TenantId = ValidTenantId })
+ "\n\n\n";
var (sessionId, tenantId, events) = NdjsonParser.ParseNdjson(ndjson);
Assert.Equal(ValidSessionId, sessionId);
Assert.Empty(events);
}
[Fact]
public void ParseNdjson_NullOrEmptyLine_IsSkipped()
{
var lines = new[]
{
JsonConvert.SerializeObject(new { SessionId = ValidSessionId, TenantId = ValidTenantId }),
"",
JsonConvert.SerializeObject(new EnrollmentEvent { EventType = "real_event" }),
};
// Note: Split('\n', RemoveEmptyEntries) already removes empty lines —
// this test verifies the parser handles extra whitespace lines gracefully.
var ndjson = string.Join('\n', lines);
var (_, _, events) = NdjsonParser.ParseNdjson(ndjson);
Assert.Single(events);
Assert.Equal("real_event", events[0].EventType);
}
[Fact]
public void ParseNdjson_EventDataWithNestedObjects_IsNormalized()
{
var lines = new[]
{
JsonConvert.SerializeObject(new { SessionId = ValidSessionId, TenantId = ValidTenantId }),
"""{"EventType":"hardware_spec","Data":{"cpuName":"Intel i7","cores":8}}""",
};
var ndjson = string.Join('\n', lines);
var (_, _, events) = NdjsonParser.ParseNdjson(ndjson);
// After normalization, values should be native .NET types, not JValue/JObject
Assert.Single(events);
var cpuName = events[0].Data["cpuName"];
Assert.IsNotType<Newtonsoft.Json.Linq.JValue>(cpuName);
Assert.Equal("Intel i7", cpuName?.ToString());
}
[Fact]
public void ParseNdjson_EmptyString_ThrowsInvalidOperationException()
{
Assert.Throws<InvalidOperationException>(() => NdjsonParser.ParseNdjson(""));
}
// --- ParseGzipAsync ---
[Fact]
public async Task ParseGzipAsync_ValidPayload_ReturnsCorrectData()
{
var payload = NdjsonParser.BuildGzipPayload(ValidSessionId, ValidTenantId,
[
new EnrollmentEvent { EventType = "phase_changed" },
new EnrollmentEvent { EventType = "app_install_started" },
]);
using var stream = new MemoryStream(payload);
var (sessionId, tenantId, events) = await NdjsonParser.ParseGzipAsync(stream, 5 * 1024 * 1024);
Assert.Equal(ValidSessionId, sessionId);
Assert.Equal(ValidTenantId, tenantId);
Assert.Equal(2, events.Count);
}
[Fact]
public async Task ParseGzipAsync_PayloadExceedsMaxSize_Throws()
{
var payload = NdjsonParser.BuildGzipPayload(ValidSessionId, ValidTenantId,
Enumerable.Range(0, 100).Select(_ => new EnrollmentEvent
{
EventType = "test",
Message = new string('x', 1000)
}));
using var stream = new MemoryStream(payload);
// Restrict to 1 byte — will throw immediately
await Assert.ThrowsAsync<InvalidOperationException>(
() => NdjsonParser.ParseGzipAsync(stream, maxSizeBytes: 1));
}
// --- BuildGzipPayload (test helper verification) ---
[Fact]
public async Task BuildGzipPayload_RoundTrip_PreservesMetadata()
{
var payload = NdjsonParser.BuildGzipPayload(ValidSessionId, ValidTenantId, []);
using var stream = new MemoryStream(payload);
var (sessionId, tenantId, events) = await NdjsonParser.ParseGzipAsync(stream, 5 * 1024 * 1024);
Assert.Equal(ValidSessionId, sessionId);
Assert.Equal(ValidTenantId, tenantId);
Assert.Empty(events);
}
}
@@ -0,0 +1,124 @@
using AutopilotMonitor.Functions.Security;
namespace AutopilotMonitor.Functions.Tests;
/// <summary>
/// Tests for SecurityValidator GUID validation.
/// StoreEventsBatchAsync relies on EnsureValidGuid to reject invalid TenantId/SessionId values —
/// these tests guard that contract.
/// </summary>
public class SecurityValidatorTests
{
// --- IsValidGuid ---
[Fact]
public void IsValidGuid_WithValidLowercaseGuid_ReturnsTrue()
{
Assert.True(SecurityValidator.IsValidGuid("a1b2c3d4-e5f6-7890-abcd-ef1234567890"));
}
[Fact]
public void IsValidGuid_WithValidUppercaseGuid_ReturnsTrue()
{
Assert.True(SecurityValidator.IsValidGuid("A1B2C3D4-E5F6-7890-ABCD-EF1234567890"));
}
[Fact]
public void IsValidGuid_WithValidMixedCaseGuid_ReturnsTrue()
{
Assert.True(SecurityValidator.IsValidGuid("a1B2c3D4-e5F6-7890-AbCd-Ef1234567890"));
}
[Fact]
public void IsValidGuid_WithNewGuid_ReturnsTrue()
{
Assert.True(SecurityValidator.IsValidGuid(Guid.NewGuid().ToString()));
}
[Fact]
public void IsValidGuid_WithNull_ReturnsFalse()
{
Assert.False(SecurityValidator.IsValidGuid(null));
}
[Fact]
public void IsValidGuid_WithEmptyString_ReturnsFalse()
{
Assert.False(SecurityValidator.IsValidGuid(""));
}
[Fact]
public void IsValidGuid_WithWhitespace_ReturnsFalse()
{
Assert.False(SecurityValidator.IsValidGuid(" "));
}
[Fact]
public void IsValidGuid_WithPlainString_ReturnsFalse()
{
Assert.False(SecurityValidator.IsValidGuid("not-a-guid"));
}
[Fact]
public void IsValidGuid_WithDeviceName_ReturnsFalse()
{
// Regression: agent running as defaultuser0 could send "DESKTOP-DIU8038\defaultuser0" as TenantId
Assert.False(SecurityValidator.IsValidGuid(@"DESKTOP-DIU8038\defaultuser0"));
}
[Fact]
public void IsValidGuid_WithGuidWithoutDashes_ReturnsFalse()
{
// No dashes → not in standard format
Assert.False(SecurityValidator.IsValidGuid("a1b2c3d4e5f67890abcdef1234567890"));
}
[Fact]
public void IsValidGuid_WithGuidInBraces_ReturnsFalse()
{
// Braced format {…} is not accepted — we require plain hyphenated form
Assert.False(SecurityValidator.IsValidGuid("{a1b2c3d4-e5f6-7890-abcd-ef1234567890}"));
}
// --- EnsureValidGuid ---
[Fact]
public void EnsureValidGuid_WithValidGuid_DoesNotThrow()
{
var ex = Record.Exception(() =>
SecurityValidator.EnsureValidGuid(Guid.NewGuid().ToString(), "TenantId"));
Assert.Null(ex);
}
[Fact]
public void EnsureValidGuid_WithNull_ThrowsArgumentException()
{
var ex = Assert.Throws<ArgumentException>(() =>
SecurityValidator.EnsureValidGuid(null, "TenantId"));
Assert.Contains("TenantId", ex.Message);
}
[Fact]
public void EnsureValidGuid_WithEmptyString_ThrowsArgumentException()
{
var ex = Assert.Throws<ArgumentException>(() =>
SecurityValidator.EnsureValidGuid("", "SessionId"));
Assert.Contains("SessionId", ex.Message);
}
[Fact]
public void EnsureValidGuid_WithInvalidString_ThrowsArgumentException()
{
var ex = Assert.Throws<ArgumentException>(() =>
SecurityValidator.EnsureValidGuid("invalid-tenant", "TenantId"));
Assert.Equal("TenantId", ex.ParamName);
}
[Fact]
public void EnsureValidGuid_ErrorMessage_MentionsParameterName()
{
var ex = Assert.Throws<ArgumentException>(() =>
SecurityValidator.EnsureValidGuid("bad", "MyParam"));
Assert.Contains("MyParam", ex.Message);
}
}
@@ -1,16 +1,12 @@
using System.IO.Compression;
using System.Linq;
using System;
using System.Net;
using System.Text;
using AutopilotMonitor.Functions.Services;
using AutopilotMonitor.Shared.Models;
using Microsoft.Azure.Functions.Worker;
using Microsoft.Azure.Functions.Worker.Http;
using Microsoft.Azure.Functions.Worker.Extensions.SignalRService;
using Microsoft.Extensions.Logging;
using Newtonsoft.Json;
using Newtonsoft.Json.Linq;
namespace AutopilotMonitor.Functions.Functions.Ingest
{
@@ -21,109 +17,24 @@ namespace AutopilotMonitor.Functions.Functions.Ingest
{
private async Task<IngestEventsRequest> ParseNdjsonGzipRequest(Stream body, string? tenantId = null)
{
// Get configuration for payload size limit (use default if tenantId not available yet)
var config = !string.IsNullOrEmpty(tenantId)
? await _configService.GetConfigurationAsync(tenantId)
: null;
var maxPayloadSizeBytes = (config?.MaxNdjsonPayloadSizeMB ?? 5) * 1024 * 1024;
// Decompress gzip with size limit protection
using var decompressed = new MemoryStream();
using (var gzip = new GZipStream(body, CompressionMode.Decompress, leaveOpen: true))
{
// Copy with size limit to prevent memory exhaustion attacks
var buffer = new byte[8192];
int bytesRead;
long totalBytesRead = 0;
while ((bytesRead = await gzip.ReadAsync(buffer, 0, buffer.Length)) > 0)
{
totalBytesRead += bytesRead;
// Check if we've exceeded the maximum payload size
if (totalBytesRead > maxPayloadSizeBytes)
{
throw new InvalidOperationException(
$"NDJSON payload size exceeds maximum allowed size of {config?.MaxNdjsonPayloadSizeMB ?? 5} MB (decompressed). " +
$"Current size: {totalBytesRead / 1024.0 / 1024.0:F2} MB"
);
}
await decompressed.WriteAsync(buffer, 0, bytesRead);
}
_logger.LogDebug($"NDJSON payload decompressed: {totalBytesRead / 1024.0:F2} KB (limit: {maxPayloadSizeBytes / 1024.0 / 1024.0} MB)");
}
decompressed.Position = 0;
var ndjson = await new StreamReader(decompressed, Encoding.UTF8).ReadToEndAsync();
// Parse NDJSON (newline-delimited JSON)
var lines = ndjson.Split('\n', StringSplitOptions.RemoveEmptyEntries);
if (lines.Length < 1)
{
throw new InvalidOperationException("NDJSON must contain at least metadata line");
}
// First line: metadata
var metadata = JsonConvert.DeserializeObject<NdjsonMetadata>(lines[0]);
if (metadata == null)
{
throw new InvalidOperationException("Failed to parse NDJSON metadata");
}
// Subsequent lines: events
var events = new List<EnrollmentEvent>();
for (int i = 1; i < lines.Length; i++)
{
var evt = JsonConvert.DeserializeObject<EnrollmentEvent>(lines[i]);
if (evt != null)
{
NormalizeEventData(evt);
events.Add(evt);
}
}
var (sessionId, metaTenantId, events) = await NdjsonParser.ParseGzipAsync(body, maxPayloadSizeBytes);
_logger.LogDebug("NDJSON payload parsed: {EventCount} events (limit: {LimitMB} MB)",
events.Count, config?.MaxNdjsonPayloadSizeMB ?? 5);
return new IngestEventsRequest
{
SessionId = metadata.SessionId,
TenantId = metadata.TenantId,
SessionId = sessionId,
TenantId = metaTenantId,
Events = events
};
}
/// <summary>
/// Normalizes event Data dictionary by converting Newtonsoft JToken objects to native .NET types.
/// Required because Newtonsoft.Json deserializes nested objects as JObject/JArray, which
/// System.Text.Json (used by SignalR) cannot serialize correctly - producing [[[]]] instead of real values.
/// </summary>
private static void NormalizeEventData(EnrollmentEvent evt)
{
if (evt.Data == null || evt.Data.Count == 0) return;
var normalized = new Dictionary<string, object>();
foreach (var kvp in evt.Data)
normalized[kvp.Key] = ConvertJTokenToNative(kvp.Value);
evt.Data = normalized;
}
private static object ConvertJTokenToNative(object value)
{
if (value is JArray jArray)
return jArray.Select(item => ConvertJTokenToNative(item)).ToList<object>();
if (value is JObject jObject)
{
var dict = new Dictionary<string, object>();
foreach (var prop in jObject.Properties())
dict[prop.Name] = ConvertJTokenToNative(prop.Value);
return dict;
}
if (value is JValue jValue)
return jValue.Value ?? string.Empty;
return value;
}
/// <summary>
/// Aggregates app install events into AppInstallSummary records
/// </summary>
@@ -121,14 +121,10 @@ namespace AutopilotMonitor.Functions.Functions.Ingest
var sessionPrefix = $"[Session: {request.SessionId.Substring(0, Math.Min(8, request.SessionId.Length))}]";
_logger.LogInformation($"{sessionPrefix} IngestEvents: Processing {request.Events.Count} events (Device: {validation.CertificateThumbprint}, Hardware: {validation.Manufacturer} {validation.Model}, Rate: {validation.RateLimitResult?.RequestsInWindow}/{validation.RateLimitResult?.MaxRequests})");
// Stamp server-side receive time, and authoritative TenantId/SessionId on all events
// Stamp server-side receive time, and authoritative TenantId/SessionId on all events.
// TenantId/SessionId come from the validated request metadata — never trust per-event agent values.
var receivedAt = DateTime.UtcNow;
foreach (var evt in request.Events)
{
evt.ReceivedAt = receivedAt;
evt.TenantId = request.TenantId;
evt.SessionId = request.SessionId;
}
StampServerFields(request.Events, request.TenantId, request.SessionId, receivedAt);
// Store events in Azure Table Storage (batch write for efficiency)
var storedEvents = await _storageService.StoreEventsBatchAsync(request.Events);
@@ -585,6 +581,23 @@ namespace AutopilotMonitor.Functions.Functions.Ingest
return new[] { summaryMessage, eventsMessage };
}
/// <summary>
/// Stamps authoritative server-side fields onto all events before storage.
/// TenantId and SessionId always come from the validated request metadata,
/// overriding any values the agent may have sent per-event.
/// Exposed as internal for unit testing.
/// </summary>
internal static void StampServerFields(
List<EnrollmentEvent> events, string tenantId, string sessionId, DateTime receivedAt)
{
foreach (var evt in events)
{
evt.ReceivedAt = receivedAt;
evt.TenantId = tenantId;
evt.SessionId = sessionId;
}
}
}
/// <summary>
@@ -0,0 +1,119 @@
using System.IO.Compression;
using System.Text;
using AutopilotMonitor.Shared.Models;
using Newtonsoft.Json;
using Newtonsoft.Json.Linq;
namespace AutopilotMonitor.Functions.Functions.Ingest
{
/// <summary>
/// Pure NDJSON+gzip parsing logic — no service dependencies.
/// Isolated here so it can be unit-tested without DI infrastructure.
/// </summary>
internal static class NdjsonParser
{
/// <summary>
/// Decompresses a gzip stream and parses NDJSON content.
/// First line: metadata (sessionId, tenantId). Subsequent lines: EnrollmentEvent objects.
/// </summary>
internal static async Task<(string sessionId, string tenantId, List<EnrollmentEvent> events)>
ParseGzipAsync(Stream body, int maxSizeBytes)
{
using var decompressed = new MemoryStream();
using (var gzip = new GZipStream(body, CompressionMode.Decompress, leaveOpen: true))
{
var buffer = new byte[8192];
int bytesRead;
long totalBytesRead = 0;
while ((bytesRead = await gzip.ReadAsync(buffer, 0, buffer.Length)) > 0)
{
totalBytesRead += bytesRead;
if (totalBytesRead > maxSizeBytes)
throw new InvalidOperationException(
$"NDJSON payload size exceeds maximum allowed size. " +
$"Current size: {totalBytesRead / 1024.0 / 1024.0:F2} MB");
await decompressed.WriteAsync(buffer, 0, bytesRead);
}
}
decompressed.Position = 0;
var ndjson = await new StreamReader(decompressed, Encoding.UTF8).ReadToEndAsync();
return ParseNdjson(ndjson);
}
/// <summary>
/// Parses a raw NDJSON string (uncompressed).
/// Exposed as internal so tests can call it directly without building gzip payloads.
/// </summary>
internal static (string sessionId, string tenantId, List<EnrollmentEvent> events)
ParseNdjson(string ndjson)
{
var lines = ndjson.Split('\n', StringSplitOptions.RemoveEmptyEntries);
if (lines.Length < 1)
throw new InvalidOperationException("NDJSON must contain at least a metadata line");
var metadata = JsonConvert.DeserializeObject<NdjsonMetadata>(lines[0]);
if (metadata == null)
throw new InvalidOperationException("Failed to parse NDJSON metadata");
var events = new List<EnrollmentEvent>();
for (int i = 1; i < lines.Length; i++)
{
var evt = JsonConvert.DeserializeObject<EnrollmentEvent>(lines[i]);
if (evt != null)
{
NormalizeEventData(evt);
events.Add(evt);
}
}
return (metadata.SessionId, metadata.TenantId, events);
}
/// <summary>
/// Builds a gzip-compressed NDJSON payload for testing.
/// </summary>
internal static byte[] BuildGzipPayload(string sessionId, string tenantId, IEnumerable<object> events)
{
var sb = new StringBuilder();
sb.AppendLine(JsonConvert.SerializeObject(new { SessionId = sessionId, TenantId = tenantId }));
foreach (var evt in events)
sb.AppendLine(JsonConvert.SerializeObject(evt));
using var output = new MemoryStream();
using (var gzip = new GZipStream(output, CompressionMode.Compress, leaveOpen: true))
{
var bytes = Encoding.UTF8.GetBytes(sb.ToString());
gzip.Write(bytes, 0, bytes.Length);
}
return output.ToArray();
}
internal static void NormalizeEventData(EnrollmentEvent evt)
{
if (evt.Data == null || evt.Data.Count == 0) return;
var normalized = new Dictionary<string, object>();
foreach (var kvp in evt.Data)
normalized[kvp.Key] = ConvertJTokenToNative(kvp.Value);
evt.Data = normalized;
}
private static object ConvertJTokenToNative(object value)
{
if (value is JArray jArray)
return jArray.Select(item => ConvertJTokenToNative(item)).ToList<object>();
if (value is JObject jObject)
{
var dict = new Dictionary<string, object>();
foreach (var prop in jObject.Properties())
dict[prop.Name] = ConvertJTokenToNative(prop.Value);
return dict;
}
if (value is JValue jValue)
return jValue.Value ?? string.Empty;
return value;
}
}
}
@@ -0,0 +1,3 @@
using System.Runtime.CompilerServices;
[assembly: InternalsVisibleTo("AutopilotMonitor.Functions.Tests")]
@@ -126,7 +126,8 @@ public static class EndpointAccessPolicyCatalog
new("PUT", "rules/analyze/{ruleId}", EndpointPolicy.TenantAdminOrGA),
new("DELETE", "rules/analyze/{ruleId}", EndpointPolicy.TenantAdminOrGA),
new("PUT", "rules/ime-log-patterns/{patternId}", EndpointPolicy.TenantAdminOrGA),
new("POST", "sessions/{sessionId}/mark-failed", EndpointPolicy.TenantAdminOrGA),
new("POST", "sessions/{sessionId}/mark-failed", EndpointPolicy.TenantAdminOrGA),
new("POST", "sessions/{sessionId}/mark-succeeded", EndpointPolicy.TenantAdminOrGA),
new("POST", "sessions/{sessionId}/report", EndpointPolicy.TenantAdminOrGA),
new("DELETE", "sessions/{sessionId}", EndpointPolicy.TenantAdminOrGA),
new("GET", "tenants/{tenantId}/admins", EndpointPolicy.TenantAdminOrGA),