[OTLP] Fix OTLP/gRPC status parsing (#7064)

Co-authored-by: martincostello <martin@martincostello.com>

Author: Kielek

src/OpenTelemetry.Exporter.OpenTelemetryProtocol/CHANGELOG.md

@@ -7,6 +7,9 @@ Notes](../../RELEASENOTES.md).
 
 ## Unreleased
 
+* Fixed an issue in OTLP/gRPC retry handling where parsing gRPC status.
+  ([#7064](https://github.com/open-telemetry/opentelemetry-dotnet/pull/7064))
+
 ## 1.15.2
 
 Released 2026-Apr-08

src/OpenTelemetry.Exporter.OpenTelemetryProtocol/Implementation/ExportClient/Grpc/GrpcStatusDeserializer.cs

@@ -125,6 +125,9 @@ internal static class GrpcStatusDeserializer
     private static Duration DecodeDuration(Stream stream)
     {
         var length = DecodeVarint(stream);
+
+        CheckLength(length, stream);
+
         var endPosition = stream.Position + length;
         long seconds = 0;
         int nanos = 0;
@@ -155,6 +158,9 @@ private static Duration DecodeDuration(Stream stream)
     private static Any DecodeAny(Stream stream)
     {
         var length = DecodeVarint(stream);
+
+        CheckLength(length, stream);
+
         var endPosition = stream.Position + length;
 
         string? typeUrl = null;
@@ -228,6 +234,9 @@ private static string DecodeString(Stream stream)
     private static byte[] DecodeBytes(Stream stream)
     {
         var length = (int)DecodeVarint(stream);
+
+        CheckLength(length, stream);
+
         var buffer = new byte[length];
         int read = stream.Read(buffer, 0, length);
         if (read != length)
@@ -250,6 +259,7 @@ private static void SkipField(Stream stream, uint wireType)
                 break;
             case WIRETYPE_LENGTH_DELIMITED:
                 var length = DecodeVarint(stream);
+                CheckLength(length, stream);
                 stream.Position += length;
                 break;
             case WIRETYPE_FIXED32:
@@ -260,6 +270,21 @@ private static void SkipField(Stream stream, uint wireType)
         }
     }
 
+    private static void CheckLength(long length, Stream stream)
+    {
+        if (length < 0 || length > int.MaxValue)
+        {
+            throw new InvalidDataException($"Invalid length: {length}.");
+        }
+
+        long available = stream.Length - stream.Position;
+
+        if (length > available)
+        {
+            throw new EndOfStreamException();
+        }
+    }
+
     internal readonly struct Duration
     {
         internal Duration(long seconds, int nanos)

test/OpenTelemetry.Exporter.OpenTelemetryProtocol.Tests/Implementation/ExportClient/GrpcStatusDeserializerTests.cs

@@ -4,6 +4,7 @@
 using Google.Protobuf;
 using Google.Protobuf.WellKnownTypes;
 using OpenTelemetry.Exporter.OpenTelemetryProtocol.Implementation.ExportClient.Grpc;
+using Type = System.Type;
 
 namespace OpenTelemetry.Exporter.OpenTelemetryProtocol.Tests.Implementation.ExportClient;
 
@@ -314,7 +315,7 @@ public void TryGetGrpcRetryDelay_OnlyNanos_ReturnsExpected()
     }
 
     [Fact]
-    public void DeserializeStatus_TruncatedStream_ThrowsEndOfStreamException()
+    public void DeserializeStatus_TruncatedStream_ThrowsException()
     {
         // Arrange: Create valid Base64 data and truncate it
         var status = new Google.Rpc.Status
@@ -332,4 +333,83 @@ public void DeserializeStatus_TruncatedStream_ThrowsEndOfStreamException()
         Assert.Throws<EndOfStreamException>(() =>
             GrpcStatusDeserializer.DeserializeStatus(grpcStatusDetailsBin));
     }
+
+    [Fact]
+    public void DeserializeStatus_WithLargeLengthDelimitedField_ThrowsException()
+    {
+        // Arrange
+        // This payload encodes a Status.details Any.value field with an extremely large
+        // length value (0x7FFFFFF0) but without enough bytes in the payload.
+        const string grpcStatusDetailsBin = "GgYS8P///wc=";
+
+        // Act & Assert
+        Assert.Throws<EndOfStreamException>(() =>
+            GrpcStatusDeserializer.DeserializeStatus(grpcStatusDetailsBin));
+    }
+
+    [Theory]
+    [InlineData("GgsS////////////AQ==")] // -1
+    [InlineData("GgYSgICAgAg=")] // 0x80000000
+    public void DeserializeStatus_WithInvalidLengthDelimitedField_ThrowsException(string grpcStatusDetailsBin)
+    {
+        // Act & Assert
+        var exception = Assert.Throws<InvalidDataException>(() => GrpcStatusDeserializer.DeserializeStatus(grpcStatusDetailsBin));
+        Assert.Contains("Invalid length", exception.Message, StringComparison.Ordinal);
+    }
+
+    [Theory]
+    [InlineData(int.MaxValue / 2, typeof(EndOfStreamException))]
+    [InlineData(int.MaxValue - 1024, typeof(EndOfStreamException))]
+    [InlineData(int.MaxValue - 1, typeof(EndOfStreamException))]
+    [InlineData(int.MaxValue, typeof(EndOfStreamException))]
+    [InlineData(uint.MaxValue, typeof(InvalidDataException))]
+    public void DeserializeStatus_InvalidDetailValueLength_Throws(long value, Type expected)
+    {
+        var anyValueLength = EncodeVarint(value);
+        var statusBytes = new byte[2 + 1 + anyValueLength.Length];
+
+        statusBytes[0] = 0x1A; // field 3 (details), wire type 2
+        statusBytes[1] = (byte)(1 + anyValueLength.Length); // embedded Any payload length
+        statusBytes[2] = 0x12; // field 2 (value), wire type 2
+        anyValueLength.CopyTo(statusBytes, 3);
+
+        var grpcStatusDetailsBin = Convert.ToBase64String(statusBytes);
+
+        Assert.Throws(expected, () => GrpcStatusDeserializer.DeserializeStatus(grpcStatusDetailsBin));
+    }
+
+    [Fact]
+    public void DeserializeStatus_InvalidEmbeddedMessageLength_Throws()
+    {
+        var statusBytes = new byte[1 + EncodeVarint(long.MaxValue).Length];
+
+        statusBytes[0] = 0x1A; // field 3 (details), wire type 2
+        EncodeVarint(long.MaxValue).CopyTo(statusBytes, 1);
+
+        var grpcStatusDetailsBin = Convert.ToBase64String(statusBytes);
+
+        Assert.Throws<InvalidDataException>(() => GrpcStatusDeserializer.DeserializeStatus(grpcStatusDetailsBin));
+    }
+
+    private static byte[] EncodeVarint(long value)
+    {
+        var encoded = new List<byte>();
+        var remaining = unchecked((ulong)value);
+
+        do
+        {
+            var current = (byte)(remaining & 0x7F);
+            remaining >>= 7;
+
+            if (remaining != 0)
+            {
+                current |= 0x80;
+            }
+
+            encoded.Add(current);
+        }
+        while (remaining != 0);
+
+        return [.. encoded];
+    }
 }

test/OpenTelemetry.Exporter.OpenTelemetryProtocol.Tests/Implementation/Transmission/OtlpExporterRetryTransmissionHandlerTests.cs

@@ -0,0 +1,117 @@
+// Copyright The OpenTelemetry Authors
+// SPDX-License-Identifier: Apache-2.0
+
+using OpenTelemetry.Exporter.OpenTelemetryProtocol.Implementation.ExportClient;
+using OpenTelemetry.Exporter.OpenTelemetryProtocol.Implementation.ExportClient.Grpc;
+
+namespace OpenTelemetry.Exporter.OpenTelemetryProtocol.Implementation.Transmission.Tests;
+
+public class OtlpExporterRetryTransmissionHandlerTests
+{
+    [Theory]
+    [InlineData(int.MaxValue / 2)]
+    [InlineData(int.MaxValue - 1024)]
+    [InlineData(int.MaxValue - 1)]
+    [InlineData(int.MaxValue)]
+    [InlineData(uint.MaxValue)]
+    public void TrySubmitRequest_FailedRequestWithOverflowingRetryHeader_LogsParsingFailureAndRetries(long value)
+    {
+        var maliciousHeader = CreateMalformedGrpcStatusDetailsHeader(value);
+        var exportClient = new RetryingTestExportClient(maliciousHeader);
+
+        using var transmissionHandler = new OtlpExporterRetryTransmissionHandler(exportClient, timeoutMilliseconds: 1_000);
+
+        bool actual;
+
+#if NET
+        using (new AllocationAssertion())
+#endif
+        {
+            actual = transmissionHandler.TrySubmitRequest([1, 2, 3], 3);
+        }
+
+        Assert.False(actual);
+        Assert.True(exportClient.SendCount >= 2, $"Expected at least 2 send attempts, but got {exportClient.SendCount}.");
+    }
+
+    private static string CreateMalformedGrpcStatusDetailsHeader(long value)
+    {
+        var anyValueLength = EncodeVarint(value);
+        var statusBytes = new byte[2 + 1 + anyValueLength.Length];
+
+        statusBytes[0] = 0x1A; // field 3 (details), wire type 2
+        statusBytes[1] = (byte)(1 + anyValueLength.Length); // embedded Any payload length
+        statusBytes[2] = 0x12; // field 2 (value), wire type 2
+
+        anyValueLength.CopyTo(statusBytes, 3);
+
+        return Convert.ToBase64String(statusBytes);
+    }
+
+    private static byte[] EncodeVarint(long value)
+    {
+        var encoded = new List<byte>();
+        var remaining = unchecked((ulong)value);
+
+        do
+        {
+            var current = (byte)(remaining & 0x7F);
+            remaining >>= 7;
+
+            if (remaining != 0)
+            {
+                current |= 0x80;
+            }
+
+            encoded.Add(current);
+        }
+        while (remaining != 0);
+
+        return [.. encoded];
+    }
+
+    private sealed class RetryingTestExportClient(string maliciousHeader) : IExportClient
+    {
+        public int SendCount { get; private set; }
+
+        public ExportClientResponse SendExportRequest(byte[] buffer, int contentLength, DateTime deadlineUtc, CancellationToken cancellationToken = default)
+        {
+            this.SendCount++;
+
+            return new ExportClientGrpcResponse(
+                success: false,
+                deadlineUtc: deadlineUtc,
+                exception: null,
+                status: new Status(StatusCode.Unavailable, "retryable"),
+                grpcStatusDetailsHeader: maliciousHeader);
+        }
+
+        public bool Shutdown(int timeoutMilliseconds) => true;
+    }
+
+#if NET
+    private sealed class AllocationAssertion : IDisposable
+    {
+        private readonly long before;
+
+        public AllocationAssertion()
+        {
+            GC.Collect();
+            GC.WaitForPendingFinalizers();
+            GC.Collect();
+
+            this.before = GC.GetTotalAllocatedBytes();
+        }
+
+        public void Dispose()
+        {
+            var allocatedBytes = GC.GetTotalAllocatedBytes() - this.before;
+
+            const int Limit = 1_000_000;
+            Assert.False(
+                allocatedBytes > Limit,
+                $"{allocatedBytes} bytes were allocated during the operation which is more than the limit of {Limit}.");
+        }
+    }
+#endif
+}