fix: Limit length of response body read to 4mb

exporter/otlp-http/lib/opentelemetry/exporter/otlp/http/trace_exporter.rb

@@ -25,6 +25,8 @@ class TraceExporter # rubocop:disable Metrics/ClassLength
           # Default timeouts in seconds.
           KEEP_ALIVE_TIMEOUT = 30
           RETRY_COUNT = 5
+          RESPONSE_BODY_LIMIT = 4_194_304 # 4 MB
+          private_constant(:KEEP_ALIVE_TIMEOUT, :RETRY_COUNT, :RESPONSE_BODY_LIMIT)
 
           ERROR_MESSAGE_INVALID_HEADERS = 'headers must be a String with comma-separated URL Encoded UTF-8 k=v pairs or a Hash'
 
@@ -145,34 +147,47 @@ def send_bytes(bytes, timeout:) # rubocop:disable Metrics/MethodLength
               @http.read_timeout = remaining_timeout
               @http.write_timeout = remaining_timeout
               @http.start unless @http.started?
-              response = @http.request(request)
-
-              case response
-              when Net::HTTPSuccess
-                response.body # Read and discard body
-                SUCCESS
-              when Net::HTTPServiceUnavailable, Net::HTTPTooManyRequests
-                response.body # Read and discard body
-                redo if backoff?(retry_after: response['Retry-After'], retry_count: retry_count += 1, reason: response.code)
-                FAILURE
-              when Net::HTTPRequestTimeOut, Net::HTTPGatewayTimeOut, Net::HTTPBadGateway
-                response.body # Read and discard body
-                redo if backoff?(retry_count: retry_count += 1, reason: response.code)
-                FAILURE
-              when Net::HTTPNotFound
-                log_request_failure(response.code)
-                FAILURE
-              when Net::HTTPBadRequest, Net::HTTPClientError, Net::HTTPServerError
-                log_status(response.body)
-                FAILURE
-              when Net::HTTPRedirection
-                @http.finish
-                handle_redirect(response['location'])
-                redo if backoff?(retry_after: 0, retry_count: retry_count += 1, reason: response.code)
-              else
-                @http.finish
-                FAILURE
+              result = nil
+              should_redo = false
+
+              measure_request_duration do # rubocop:disable Metrics/BlockLength
+                @http.request(request) do |response| # rubocop:disable Metrics/BlockLength
+                  case response
+                  when Net::HTTPSuccess
+                    response.read_body { |_| } # Drain and discard, preserves keep-alive
+                    result = SUCCESS
+                  when Net::HTTPServiceUnavailable, Net::HTTPTooManyRequests
+                    response.read_body { |_| }
+                    should_redo = backoff?(retry_after: response['Retry-After'], retry_count: retry_count += 1, reason: response.code)
+                    result = FAILURE
+                  when Net::HTTPRequestTimeOut, Net::HTTPGatewayTimeOut, Net::HTTPBadGateway
+                    response.read_body { |_| }
+                    should_redo = backoff?(retry_count: retry_count += 1, reason: response.code)
+                    result = FAILURE
+                  when Net::HTTPNotFound
+                    log_request_failure(response.code)
+                    FAILURE
+                  when Net::HTTPBadRequest, Net::HTTPClientError, Net::HTTPServerError
+                    body, truncated = read_response_body(response)
+                    log_status(body, truncated: truncated)
+                    @metrics_reporter.add_to_counter('otel.otlp_exporter.failure', labels: { 'reason' => response.code })
+                    result = FAILURE
+                  when Net::HTTPRedirection
+                    response.read_body { |_| }
+                    @http.finish
+                    handle_redirect(response['location'])
+                    should_redo = backoff?(retry_after: 0, retry_count: retry_count += 1, reason: response.code)
+                  else
+                    response.read_body { |_| }
+                    @http.finish
+                    result = FAILURE
+                  end
+                end
               end
+
+              redo if should_redo
+
+              result
             rescue Net::OpenTimeout, Net::ReadTimeout
               retry if backoff?(retry_count: retry_count += 1, reason: 'timeout')
               return FAILURE
@@ -207,7 +222,13 @@ def handle_redirect(location)
             # TODO: figure out destination and reinitialize @http and @path
           end
 
-          def log_status(body)
+          def log_status(body, truncated: false)
+            if truncated
+              OpenTelemetry.handle_error(message: "OTLP exporter received an oversized error response body (truncated at #{RESPONSE_BODY_LIMIT} bytes)")
+              return
+            end
+            return if body.nil? || body.empty?
+
             status = Google::Rpc::Status.decode(body)
             pool = ::Google::Protobuf::DescriptorPool.generated_pool
             details = status.details.filter_map do |detail|
@@ -219,6 +240,54 @@ def log_status(body)
             OpenTelemetry.handle_error(exception: e, message: 'unexpected error decoding rpc.Status in OTLP::Exporter#log_status')
           end
 
+          def read_response_body(response) # rubocop:disable Metrics/MethodLength
+            return ['', false] if response.nil?
+
+            content_length = response['content-length']&.to_i
+            if content_length && content_length > RESPONSE_BODY_LIMIT
+              @http.finish # closes socket without reading any of the oversized body
+              return ['', true]
+            end
+
+            body = +''
+            truncated = false
+
+            response.read_body do |chunk|
+              remaining = RESPONSE_BODY_LIMIT - body.bytesize
+              body << chunk.byteslice(0, remaining)
+
+              if chunk.bytesize > remaining
+                truncated = true
+                @http.finish # closes socket, nil's the body or else net/http will attempt to read the rest of the response
+                break
+              end
+            end
+
+            body.force_encoding('UTF-8')
+            body.scrub! if truncated # truncation may have split a multi-byte character
+            [body, truncated]
+          rescue IOError
+            raise unless truncated # we'll handle this when we know net/http is upset trying to read after http.finish
+
+            [body || '', truncated]
+          rescue StandardError => e
+            OpenTelemetry.handle_error(exception: e, message: 'error reading response body')
+            ['', false]
+          end
+
+          def measure_request_duration
+            start = Process.clock_gettime(Process::CLOCK_MONOTONIC)
+            begin
+              response = yield
+            ensure
+              stop = Process.clock_gettime(Process::CLOCK_MONOTONIC)
+              duration_ms = 1000.0 * (stop - start)
+              @metrics_reporter.record_value('otel.otlp_exporter.request_duration',
+                                             value: duration_ms,
+                                             labels: { 'status' => response&.code || 'unknown' })
+            end
+          end
+
           def log_request_failure(response_code)
             OpenTelemetry.handle_error(message: "OTLP exporter received http.code=#{response_code} for uri='#{@uri}' in OTLP::Exporter#send_bytes")
           end

exporter/otlp-http/test/opentelemetry/exporter/otlp/http/trace_exporter_test.rb

@@ -846,4 +846,232 @@
       end
     end
   end
+
+  describe 'response body reading' do
+    let(:exporter) { OpenTelemetry::Exporter::OTLP::HTTP::TraceExporter.new }
+    let(:span_data) { OpenTelemetry::TestHelpers.create_span_data }
+
+    it 'discards body for successful responses without reading into memory' do
+      stub_request(:post, 'http://localhost:4318/v1/traces').to_return(status: 200, body: 'success body')
+
+      result = exporter.export([span_data])
+
+      _(result).must_equal(success)
+    end
+
+    it 'discards body for retryable responses without reading into memory' do
+      stub_request(:post, 'http://localhost:4318/v1/traces')
+        .to_return(status: 503, body: 'service unavailable', headers: { 'Retry-After' => '0' })
+        .then.to_return(status: 200)
+
+      result = exporter.export([span_data])
+
+      _(result).must_equal(success)
+    end
+
+    it 'reads and parses error response body smaller than limit' do
+      log_stream = StringIO.new
+      logger = OpenTelemetry.logger
+      OpenTelemetry.logger = ::Logger.new(log_stream)
+
+      details = [::Google::Protobuf::Any.pack(::Google::Protobuf::StringValue.new(value: 'error details'))]
+      status = ::Google::Rpc::Status.encode(::Google::Rpc::Status.new(code: 3, message: 'invalid argument', details: details))
+      stub_request(:post, 'http://localhost:4318/v1/traces').to_return(status: 400, body: status)
+
+      result = exporter.export([span_data])
+
+      _(result).must_equal(export_failure)
+      _(log_stream.string).must_match(/invalid argument/)
+      _(log_stream.string).wont_match(/truncated/)
+    ensure
+      OpenTelemetry.logger = logger
+    end
+
+    it 'truncates error response body larger than 4 MB limit' do
+      log_stream = StringIO.new
+      logger = OpenTelemetry.logger
+      OpenTelemetry.logger = ::Logger.new(log_stream)
+
+      # Create a body larger than 4 MB
+      large_message = 'x' * 5_000_000 # 5 MB
+      details = [::Google::Protobuf::Any.pack(::Google::Protobuf::StringValue.new(value: large_message))]
+      large_status = ::Google::Rpc::Status.new(code: 3, message: 'large error', details: details)
+      large_body = ::Google::Rpc::Status.encode(large_status)
+
+      stub_request(:post, 'http://localhost:4318/v1/traces').to_return(status: 400, body: large_body)
+
+      result = exporter.export([span_data])
+
+      _(result).must_equal(export_failure)
+      _(log_stream.string).must_match(/oversized error response body/)
+    ensure
+      OpenTelemetry.logger = logger
+    end
+
+    it 'handles malformed error response body gracefully' do
+      log_stream = StringIO.new
+      logger = OpenTelemetry.logger
+      OpenTelemetry.logger = ::Logger.new(log_stream)
+
+      stub_request(:post, 'http://localhost:4318/v1/traces').to_return(status: 400, body: 'not valid protobuf')
+
+      result = exporter.export([span_data])
+
+      _(result).must_equal(export_failure)
+      _(log_stream.string).must_match(/unexpected error decoding rpc.Status/)
+    ensure
+      OpenTelemetry.logger = logger
+    end
+  end
+
+  describe 'response body reading with real sockets' do
+    # These tests use a real TCPServer instead of WebMock to verify
+    # behavior against actual Net::HTTP socket I/O. WebMock's patching
+    # of Net::HTTP changes how read_body works, so it doesn't exercise
+    # the same code paths as real Net::HTTP.
+
+    def with_fake_server(response_body_size: nil, status: 400, handler: nil)
+      server = TCPServer.new('127.0.0.1', 0)
+      port = server.addr[1]
+      handler ||= ->(srv, stat) { handle_fake_request(srv, 'X' * response_body_size, stat) }
+
+      server_thread = Thread.new { handler.call(server, status) }
+
+      WebMock::HttpLibAdapters::NetHttpAdapter.disable!
+      yield port
+    ensure
+      WebMock::HttpLibAdapters::NetHttpAdapter.enable!
+      server&.close
+      server_thread&.join(2)
+    end
+
+    def handle_fake_request(server, body, status)
+      client = server.accept
+      content_length = read_content_length(client)
+      client.read(content_length) if content_length > 0
+
+      client.print "HTTP/1.1 #{status} Bad Request\r\n"
+      client.print "Content-Type: application/x-protobuf\r\n"
+      client.print "Content-Length: #{body.bytesize}\r\n"
+      client.print "Connection: close\r\n"
+      client.print "\r\n"
+      client.write body
+      client.close
+    rescue StandardError
+      # client may disconnect early
+    end
+
+    # Sends only headers announcing an oversized Content-Length, then closes
+    # without ever writing a body — proves the exporter short-circuits on
+    # Content-Length rather than attempting to stream-read the body.
+    def handle_fake_headers_only_request(server, declared_content_length, status)
+      client = server.accept
+      content_length = read_content_length(client)
+      client.read(content_length) if content_length > 0
+
+      client.print "HTTP/1.1 #{status} Bad Request\r\n"
+      client.print "Content-Type: application/x-protobuf\r\n"
+      client.print "Content-Length: #{declared_content_length}\r\n"
+      client.print "Connection: close\r\n"
+      client.print "\r\n"
+      client.close
+    rescue StandardError
+      # client may disconnect early
+    end
+
+    # Writes body using chunked transfer-encoding with no Content-Length
+    # header, the common shape of a real collector's error response.
+    def handle_fake_chunked_request(server, body, status)
+      client = server.accept
+      content_length = read_content_length(client)
+      client.read(content_length) if content_length > 0
+
+      client.print "HTTP/1.1 #{status} Bad Request\r\n"
+      client.print "Content-Type: application/x-protobuf\r\n"
+      client.print "Transfer-Encoding: chunked\r\n"
+      client.print "Connection: close\r\n"
+      client.print "\r\n"
+
+      chunk_size = 65_536
+      offset = 0
+      while offset < body.bytesize
+        chunk = body.byteslice(offset, chunk_size)
+        client.print "#{chunk.bytesize.to_s(16)}\r\n#{chunk}\r\n"
+        offset += chunk_size
+      end
+      client.print "0\r\n\r\n"
+      client.close
+    rescue StandardError
+      # client may disconnect early
+    end
+
+    def read_content_length(client)
+      content_length = 0
+      while (line = client.gets) && line != "\r\n"
+        content_length = line.split(': ', 2).last.to_i if line.start_with?('Content-Length')
+      end
+      content_length
+    end
+
+    it 'limits error response body read to 4 MB against a real HTTP server' do
+      log_stream = StringIO.new
+      logger = OpenTelemetry.logger
+      OpenTelemetry.logger = ::Logger.new(log_stream)
+
+      with_fake_server(response_body_size: 5_000_000) do |port|
+        exporter = OpenTelemetry::Exporter::OTLP::HTTP::TraceExporter.new(
+          endpoint: "http://127.0.0.1:#{port}/v1/traces"
+        )
+        result = exporter.export([OpenTelemetry::TestHelpers.create_span_data])
+
+        _(result).must_equal(export_failure)
+        _(log_stream.string).must_match(/oversized error response body/)
+        _(log_stream.string).wont_match(/unexpected error decoding/)
+        _(log_stream.string).wont_match(/read_body called twice/)
+      end
+    ensure
+      OpenTelemetry.logger = logger
+    end
+
+    it 'skips reading the body when Content-Length already exceeds the limit' do
+      log_stream = StringIO.new
+      logger = OpenTelemetry.logger
+      OpenTelemetry.logger = ::Logger.new(log_stream)
+
+      handler = ->(server, status) { handle_fake_headers_only_request(server, 5_000_000, status) }
+      with_fake_server(handler: handler) do |port|
+        exporter = OpenTelemetry::Exporter::OTLP::HTTP::TraceExporter.new(
+          endpoint: "http://127.0.0.1:#{port}/v1/traces"
+        )
+        result = exporter.export([OpenTelemetry::TestHelpers.create_span_data])
+
+        _(result).must_equal(export_failure)
+        _(log_stream.string).must_match(/oversized error response body/)
+        _(log_stream.string).wont_match(/error reading response body/)
+      end
+    ensure
+      OpenTelemetry.logger = logger
+    end
+
+    it 'limits chunked (no Content-Length) response bodies to 4 MB' do
+      log_stream = StringIO.new
+      logger = OpenTelemetry.logger
+      OpenTelemetry.logger = ::Logger.new(log_stream)
+
+      large_body = 'X' * 5_000_000
+      handler = ->(server, status) { handle_fake_chunked_request(server, large_body, status) }
+      with_fake_server(handler: handler) do |port|
+        exporter = OpenTelemetry::Exporter::OTLP::HTTP::TraceExporter.new(
+          endpoint: "http://127.0.0.1:#{port}/v1/traces"
+        )
+        result = exporter.export([OpenTelemetry::TestHelpers.create_span_data])
+
+        _(result).must_equal(export_failure)
+        _(log_stream.string).must_match(/oversized error response body/)
+        _(log_stream.string).wont_match(/unexpected error decoding/)
+      end
+    ensure
+      OpenTelemetry.logger = logger
+    end
+  end
 end