From c129ad0ed44692fd34cd17b47264f3aea75a8cd8 Mon Sep 17 00:00:00 2001 From: Hiroshi Hatake Date: Tue, 20 Oct 2026 16:23:20 +0900 Subject: [PATCH 1/4] snappy: consolidate framed decoding with alignment-safe checksum reads Signed-off-by: Hiroshi Hatake --- include/fluent-bit/flb_snappy.h | 4 + src/flb_snappy.c | 330 ++++++++++---------------------- 2 files changed, 104 insertions(+), 230 deletions(-) diff --git a/include/fluent-bit/flb_snappy.h b/include/fluent-bit/flb_snappy.h index e1e34e88be6..d419ae6dc3c 100644 --- a/include/fluent-bit/flb_snappy.h +++ b/include/fluent-bit/flb_snappy.h @@ -54,4 +54,8 @@ int flb_snappy_uncompress(char *in_data, size_t in_len, int flb_snappy_uncompress_framed_data(char *in_data, size_t in_len, char **out_data, size_t *out_len); +/* Decode raw or framed Snappy within limit; returns FLB_DECOMPRESSOR_* status. */ +int flb_snappy_uncompress_bounded(char **out_data, size_t *out_len, + char *in_data, size_t in_len, size_t limit); + #endif diff --git a/src/flb_snappy.c b/src/flb_snappy.c index 59da06fda16..d757e7bb197 100644 --- a/src/flb_snappy.c +++ b/src/flb_snappy.c @@ -26,6 +26,7 @@ #include #include +#include #include @@ -106,258 +107,127 @@ int flb_snappy_uncompress(char *in_data, size_t in_len, return 0; } -static uint32_t calculate_checksum(char *buffer, size_t length) +static uint32_t snappy_read_le32(const unsigned char *buffer) { - uint32_t checksum; - - checksum = cfl_checksum_crc32c((unsigned char *) buffer, length); - - return ((checksum >> 15) | - (checksum << 17)) + 0xa282ead8; + /* Framed checksums are little-endian and may start at any byte alignment. */ + return ((uint32_t) buffer[0]) | ((uint32_t) buffer[1] << 8) | + ((uint32_t) buffer[2] << 16) | ((uint32_t) buffer[3] << 24); } -int flb_snappy_uncompress_framed_data(char *in_data, size_t in_len, - char **out_data, size_t *out_len) +int flb_snappy_uncompress_bounded(char **output_buffer, size_t *output_size, + char *input_buffer, size_t input_size, size_t limit) { - uint32_t decompressed_data_checksum; - size_t stream_identifier_length; - size_t uncompressed_chunk_count; - int stream_identifier_found; - char *aggregated_data_buffer; - size_t aggregated_data_length = 0; - size_t aggregated_data_offset; - size_t compressed_chunk_count; - struct cfl_list *iterator_backup; - uint32_t frame_checksum; - char *frame_buffer; - size_t frame_length; - char *frame_body; - unsigned char frame_type; - struct cfl_list *iterator; - int result; - size_t offset; - struct cfl_list chunks; - struct flb_snappy_data_chunk *chunk; - - if (*((uint8_t *) in_data) != FLB_SNAPPY_FRAME_TYPE_STREAM_IDENTIFIER) { - return flb_snappy_uncompress(in_data, in_len, out_data, out_len); - } + size_t offset; + size_t length; + size_t expanded; + size_t total = 0; + size_t output_offset = 0; + int pass; + unsigned char type; + unsigned char *frame; + uint32_t checksum; + char *buffer = NULL; - if (out_data == NULL) { - return -1; + if (output_buffer == NULL || output_size == NULL) { + return FLB_DECOMPRESSOR_FAILURE; } - - if (out_len == NULL) { - return -1; + *output_buffer = NULL; + *output_size = 0; + if (input_buffer == NULL || input_size == 0) { + return FLB_DECOMPRESSOR_FAILURE; } - - *out_data = NULL; - *out_len = 0; - - cfl_list_init(&chunks); - - compressed_chunk_count = 0; - uncompressed_chunk_count = 0; - - stream_identifier_found = FLB_FALSE; - stream_identifier_length = strlen(FLB_SNAPPY_STREAM_IDENTIFIER_STRING); - - result = 0; - offset = 0; - - while (offset < in_len && result == 0) { - if (offset + 4 > in_len) { - result = -2; - break; - } - - frame_buffer = &in_data[offset]; - - frame_type = *((uint8_t *) &frame_buffer[0]); - - frame_length = ((uint32_t)((unsigned char) frame_buffer[1])) | - ((uint32_t)((unsigned char) frame_buffer[2]) << 8) | - ((uint32_t)((unsigned char) frame_buffer[3]) << 16); - - frame_body = &frame_buffer[4]; - - if (frame_length > FLB_SNAPPY_FRAME_SIZE_LIMIT) { - result = -2; + if (input_size < 10 || + (unsigned char) input_buffer[0] != FLB_SNAPPY_FRAME_TYPE_STREAM_IDENTIFIER || + input_buffer[1] != 6 || input_buffer[2] != 0 || input_buffer[3] != 0 || + memcmp(input_buffer + 4, FLB_SNAPPY_STREAM_IDENTIFIER_STRING, 6) != 0) { + if (!snappy_uncompressed_length(input_buffer, input_size, &expanded)) { + return FLB_DECOMPRESSOR_FAILURE; } - else if (offset + 4 + frame_length > in_len) { - result = -2; + if (expanded > limit) { + return FLB_DECOMPRESSOR_LIMIT_EXCEEDED; } - else if (frame_type == FLB_SNAPPY_FRAME_TYPE_STREAM_IDENTIFIER) { - if (!stream_identifier_found) { - if (frame_length == stream_identifier_length) { - result = strncmp(frame_body, - FLB_SNAPPY_STREAM_IDENTIFIER_STRING, - stream_identifier_length); - - if (result == 0) { - stream_identifier_found = FLB_TRUE; - } - } + return flb_snappy_uncompress(input_buffer, input_size, output_buffer, output_size) == 0 ? + 0 : FLB_DECOMPRESSOR_FAILURE; + } + + /* Inspect every framed chunk before allocating a single bounded output. */ + for (pass = 0; pass < 2; pass++) { + offset = 0; + while (offset < input_size) { + if (input_size - offset < 4) { + goto invalid; } - } - else if (frame_type == FLB_SNAPPY_FRAME_TYPE_COMPRESSED_DATA) { - if (frame_length < sizeof(uint32_t)) { - result = -2; - break; + frame = (unsigned char *) input_buffer + offset; + type = frame[0]; + length = ((size_t) frame[1]) | ((size_t) frame[2] << 8) | + ((size_t) frame[3] << 16); + if (length > input_size - offset - 4) { + goto invalid; } - - chunk = (struct flb_snappy_data_chunk * ) \ - flb_calloc(1, sizeof(struct flb_snappy_data_chunk)); - - if (chunk != NULL) { - /* We add the chunk to the list now because that way - * even if the process fails we can clean up in a single - * place. - */ - compressed_chunk_count++; - - chunk->dynamically_allocated_buffer = FLB_TRUE; - - cfl_list_add(&chunk->_head, &chunks); - - frame_checksum = *((uint32_t *) &frame_body[0]); - frame_body = &frame_body[4]; - - result = flb_snappy_uncompress( - frame_body, - frame_length - sizeof(uint32_t), - &chunk->buffer, - &chunk->length); - - /* decompressed data */ - if (result == 0) { - decompressed_data_checksum = calculate_checksum( - chunk->buffer, - chunk->length); - - if (decompressed_data_checksum != frame_checksum) { - result = -3; - } - else { - aggregated_data_length += chunk->length; - } - } - else { - result = -4; + frame += 4; + if (type == FLB_SNAPPY_FRAME_TYPE_STREAM_IDENTIFIER) { + if (length != 6 || memcmp(frame, FLB_SNAPPY_STREAM_IDENTIFIER_STRING, 6) != 0) { + goto invalid; } } - } - else if (frame_type == FLB_SNAPPY_FRAME_TYPE_UNCOMPRESSED_DATA) { - if (frame_length < sizeof(uint32_t)) { - result = -2; - break; - } - - chunk = (struct flb_snappy_data_chunk *) \ - flb_calloc(1, sizeof(struct flb_snappy_data_chunk)); - - if (chunk != NULL) { - /* We add the chunk to the list now because that way - * even if the process fails we can clean up in a single - * place. - */ - uncompressed_chunk_count++; - - chunk->dynamically_allocated_buffer = FLB_FALSE; - - cfl_list_add(&chunk->_head, &chunks); - - frame_checksum = *((uint32_t *) &frame_body[0]); - frame_body = &frame_body[4]; - - chunk->buffer = frame_body; - chunk->length = frame_length - sizeof(uint32_t); - - decompressed_data_checksum = calculate_checksum( - chunk->buffer, - chunk->length); - - if (decompressed_data_checksum != frame_checksum) { - result = -3; + else if (type == FLB_SNAPPY_FRAME_TYPE_COMPRESSED_DATA || + type == FLB_SNAPPY_FRAME_TYPE_UNCOMPRESSED_DATA) { + if (length < 4 || length > FLB_SNAPPY_FRAME_SIZE_LIMIT) { + goto invalid; + } + expanded = length - 4; + if (type == FLB_SNAPPY_FRAME_TYPE_COMPRESSED_DATA && + !snappy_uncompressed_length((char *) frame + 4, length - 4, &expanded)) { + goto invalid; + } + if (pass == 0) { + if (expanded > limit - total) { + return FLB_DECOMPRESSOR_LIMIT_EXCEEDED; + } + total += expanded; } else { - aggregated_data_length += chunk->length; + if (type == FLB_SNAPPY_FRAME_TYPE_COMPRESSED_DATA) { + if (snappy_uncompress((char *) frame + 4, length - 4, + buffer + output_offset) != 0) { + goto invalid; + } + } + else { + memcpy(buffer + output_offset, frame + 4, expanded); + } + checksum = cfl_checksum_crc32c((unsigned char *) buffer + output_offset, + expanded); + checksum = ((checksum >> 15) | (checksum << 17)) + 0xa282ead8; + if (checksum != snappy_read_le32(frame)) { + goto invalid; + } + output_offset += expanded; } } - } - else if (frame_type == FLB_SNAPPY_FRAME_TYPE_PADDING) { - /* We just need to skip these frames */ - } - else if (frame_type >= FLB_SNAPPY_FRAME_TYPE_RESERVED_UNSKIPPABLE_BASE && - frame_type <= FLB_SNAPPY_FRAME_TYPE_RESERVED_UNSKIPPABLE_TOP) { - result = -5; - } - else if (frame_type >= FLB_SNAPPY_FRAME_TYPE_RESERVED_SKIPPABLE_BASE && - frame_type <= FLB_SNAPPY_FRAME_TYPE_RESERVED_SKIPPABLE_TOP) { - /* We just need to skip these frames */ - } - - offset += frame_length + 4; - } - - aggregated_data_buffer = NULL; - aggregated_data_length = 0; - - if (compressed_chunk_count == 1 && - uncompressed_chunk_count == 0 && - result == 0) { - /* This is a "past path" to avoid unnecessarily copying - * data whene the input is only comprised of a single - * compressed chunk. - */ - - chunk = cfl_list_entry_first(&chunks, - struct flb_snappy_data_chunk, _head); - - aggregated_data_buffer = chunk->buffer; - aggregated_data_length = chunk->length; - aggregated_data_offset = aggregated_data_length; - - flb_free(chunk); - } - else { - if (aggregated_data_length > 0) { - aggregated_data_buffer = flb_calloc(aggregated_data_length, - sizeof(char)); - - if (aggregated_data_buffer == NULL) { - result = -6; + else if (type < FLB_SNAPPY_FRAME_TYPE_RESERVED_SKIPPABLE_BASE) { + goto invalid; } + offset += 4 + length; } - - aggregated_data_offset = 0; - cfl_list_foreach_safe(iterator, iterator_backup, &chunks) { - chunk = cfl_list_entry(iterator, - struct flb_snappy_data_chunk, _head); - - if (chunk->buffer != NULL) { - if (aggregated_data_buffer != NULL && - result == 0) { - memcpy(&aggregated_data_buffer[aggregated_data_offset], - chunk->buffer, - chunk->length); - - aggregated_data_offset += chunk->length; - } - - if (chunk->dynamically_allocated_buffer) { - flb_free(chunk->buffer); - } + if (pass == 0) { + buffer = flb_malloc(total == 0 ? 1 : total); + if (buffer == NULL) { + return FLB_DECOMPRESSOR_FAILURE; } - - cfl_list_del(&chunk->_head); - - flb_free(chunk); } } + *output_buffer = buffer; + *output_size = total; + return 0; - *out_data = (char *) aggregated_data_buffer; - *out_len = aggregated_data_offset; +invalid: + flb_free(buffer); + return FLB_DECOMPRESSOR_FAILURE; +} - return result; +int flb_snappy_uncompress_framed_data(char *in_data, size_t in_len, + char **out_data, size_t *out_len) +{ + return flb_snappy_uncompress_bounded(out_data, out_len, in_data, in_len, SIZE_MAX); } From 3a5168d381376311b7553505fb9402e527f20c44 Mon Sep 17 00:00:00 2001 From: Hiroshi Hatake Date: Tue, 20 Oct 2026 16:23:36 +0900 Subject: [PATCH 2/4] compression: reuse the shared bounded Snappy decoder Signed-off-by: Hiroshi Hatake --- src/flb_compression.c | 117 +----------------------------------------- 1 file changed, 1 insertion(+), 116 deletions(-) diff --git a/src/flb_compression.c b/src/flb_compression.c index 4775ab2cd56..987fd8b4dac 100644 --- a/src/flb_compression.c +++ b/src/flb_compression.c @@ -24,8 +24,6 @@ #include #include #include -#include -#include struct flb_snappy_decompression_context { char *buffer; @@ -305,119 +303,6 @@ int flb_decompress(struct flb_decompression_context *context, return FLB_DECOMPRESSOR_SUCCESS; } -static uint32_t decompression_read_le32(const unsigned char *buffer) -{ - return ((uint32_t) buffer[0]) | ((uint32_t) buffer[1] << 8) | - ((uint32_t) buffer[2] << 16) | ((uint32_t) buffer[3] << 24); -} - -static int decompress_snappy_bounded(char **output_buffer, size_t *output_size, - char *input_buffer, size_t input_size, size_t limit) -{ - size_t offset; - size_t length; - size_t expanded; - size_t total = 0; - size_t output_offset = 0; - int pass; - unsigned char type; - unsigned char *frame; - uint32_t checksum; - char *buffer = NULL; - - if (input_size == 0) { - return FLB_DECOMPRESSOR_FAILURE; - } - if (input_size < 10 || - (unsigned char) input_buffer[0] != FLB_SNAPPY_FRAME_TYPE_STREAM_IDENTIFIER || - input_buffer[1] != 6 || input_buffer[2] != 0 || input_buffer[3] != 0 || - memcmp(input_buffer + 4, FLB_SNAPPY_STREAM_IDENTIFIER_STRING, 6) != 0) { - if (!snappy_uncompressed_length(input_buffer, input_size, &expanded)) { - return FLB_DECOMPRESSOR_FAILURE; - } - if (expanded > limit) { - return FLB_DECOMPRESSOR_LIMIT_EXCEEDED; - } - return flb_snappy_uncompress(input_buffer, input_size, output_buffer, output_size) == 0 ? - 0 : FLB_DECOMPRESSOR_FAILURE; - } - - /* Inspect every framed chunk before allocating a single bounded output. */ - for (pass = 0; pass < 2; pass++) { - offset = 0; - while (offset < input_size) { - if (input_size - offset < 4) { - goto invalid; - } - frame = (unsigned char *) input_buffer + offset; - type = frame[0]; - length = ((size_t) frame[1]) | ((size_t) frame[2] << 8) | - ((size_t) frame[3] << 16); - if (length > input_size - offset - 4) { - goto invalid; - } - frame += 4; - if (type == FLB_SNAPPY_FRAME_TYPE_STREAM_IDENTIFIER) { - if (length != 6 || memcmp(frame, FLB_SNAPPY_STREAM_IDENTIFIER_STRING, 6) != 0) { - goto invalid; - } - } - else if (type == FLB_SNAPPY_FRAME_TYPE_COMPRESSED_DATA || - type == FLB_SNAPPY_FRAME_TYPE_UNCOMPRESSED_DATA) { - if (length < 4 || length > FLB_SNAPPY_FRAME_SIZE_LIMIT) { - goto invalid; - } - expanded = length - 4; - if (type == FLB_SNAPPY_FRAME_TYPE_COMPRESSED_DATA && - !snappy_uncompressed_length((char *) frame + 4, length - 4, &expanded)) { - goto invalid; - } - if (pass == 0) { - if (expanded > limit - total) { - return FLB_DECOMPRESSOR_LIMIT_EXCEEDED; - } - total += expanded; - } - else { - if (type == FLB_SNAPPY_FRAME_TYPE_COMPRESSED_DATA) { - if (snappy_uncompress((char *) frame + 4, length - 4, - buffer + output_offset) != 0) { - goto invalid; - } - } - else { - memcpy(buffer + output_offset, frame + 4, expanded); - } - checksum = cfl_checksum_crc32c((unsigned char *) buffer + output_offset, - expanded); - checksum = ((checksum >> 15) | (checksum << 17)) + 0xa282ead8; - if (checksum != decompression_read_le32(frame)) { - goto invalid; - } - output_offset += expanded; - } - } - else if (type < FLB_SNAPPY_FRAME_TYPE_RESERVED_SKIPPABLE_BASE) { - goto invalid; - } - offset += 4 + length; - } - if (pass == 0) { - buffer = flb_malloc(total == 0 ? 1 : total); - if (buffer == NULL) { - return FLB_DECOMPRESSOR_FAILURE; - } - } - } - *output_buffer = buffer; - *output_size = total; - return 0; - -invalid: - flb_free(buffer); - return FLB_DECOMPRESSOR_FAILURE; -} - static int decompress_snappy_dispatch(struct flb_decompression_context *context, void *output_buffer, size_t *output_size) { @@ -427,7 +312,7 @@ static int decompress_snappy_dispatch(struct flb_decompression_context *context, *output_size = 0; if (snappy->buffer == NULL) { - ret = decompress_snappy_bounded(&snappy->buffer, &snappy->length, + ret = flb_snappy_uncompress_bounded(&snappy->buffer, &snappy->length, (char *) context->read_buffer, context->input_buffer_length, context->output_limit - context->output_size); From c126a13f3e2b54075374dc834579a35bf93fadaf Mon Sep 17 00:00:00 2001 From: Hiroshi Hatake Date: Tue, 20 Oct 2026 16:23:37 +0900 Subject: [PATCH 3/4] tests: cover Snappy frame alignment and malformed checksums Signed-off-by: Hiroshi Hatake --- tests/internal/compression.c | 75 ++++++++++++++++++++++++++++++++++++ 1 file changed, 75 insertions(+) diff --git a/tests/internal/compression.c b/tests/internal/compression.c index b8e24163405..6ae677fc473 100644 --- a/tests/internal/compression.c +++ b/tests/internal/compression.c @@ -144,6 +144,80 @@ static void test_snappy_framed_checksum(void) TEST_CHECK(output == NULL && output_size == 0); } +static void test_snappy_legacy_framed(void) +{ + const unsigned char frames[][23] = { + {0xff, 6, 0, 0, 's', 'N', 'a', 'P', 'p', 'Y', + 1, 7, 0, 0, 0x6e, 0x57, 0xf1, 0x21, 'a', 'b', 'c'}, + {0xff, 6, 0, 0, 's', 'N', 'a', 'P', 'p', 'Y', + 0, 9, 0, 0, 0x6e, 0x57, 0xf1, 0x21, 3, 8, 'a', 'b', 'c'} + }; + const size_t lengths[] = {21, 23}; + unsigned char *storage; + unsigned char *input; + char *output; + size_t output_size; + size_t alignment; + size_t kind; + size_t length; + int ret; + + storage = flb_malloc(64); + TEST_ASSERT(storage != NULL); + for (kind = 0; kind < 2; kind++) { + for (alignment = 0; alignment < 4; alignment++) { + input = storage + alignment; + length = lengths[kind]; + memcpy(input, frames[kind], length); + /* Fixed wire bytes verify little-endian CRCs at every alignment. */ + ret = flb_snappy_uncompress_framed_data((char *) input, length, + &output, &output_size); + TEST_CHECK(ret == 0); + TEST_CHECK(output_size == 3); + if (output != NULL && output_size == 3) { + TEST_CHECK(memcmp(output, "abc", 3) == 0); + } + flb_free(output); + + input[14] ^= 1; + ret = flb_snappy_uncompress_framed_data((char *) input, length, + &output, &output_size); + TEST_CHECK(ret != 0); + TEST_CHECK(output == NULL && output_size == 0); + input[14] ^= 1; + ret = flb_snappy_uncompress_framed_data((char *) input, length - 1, + &output, &output_size); + TEST_CHECK(ret != 0); + TEST_CHECK(output == NULL && output_size == 0); + + /* Append the other frame type, without a second identifier. */ + memcpy(input + length, frames[1 - kind] + 10, lengths[1 - kind] - 10); + ret = flb_snappy_uncompress_framed_data((char *) input, 34, + &output, &output_size); + TEST_CHECK(ret == 0); + TEST_CHECK(output_size == 6); + if (output != NULL && output_size == 6) { + TEST_CHECK(memcmp(output, "abcabc", 6) == 0); + } + flb_free(output); + } + } + /* The report's minimal frame and a checksum shorter than four bytes. */ + memcpy(storage, frames[0], 18); + storage[11] = 4; + ret = flb_snappy_uncompress_framed_data((char *) storage, 18, &output, &output_size); + TEST_CHECK(ret != 0); + TEST_CHECK(output == NULL && output_size == 0); + storage[11] = 3; + ret = flb_snappy_uncompress_framed_data((char *) storage, 17, &output, &output_size); + TEST_CHECK(ret != 0); + TEST_CHECK(output == NULL && output_size == 0); + ret = flb_snappy_uncompress_framed_data(NULL, 0, &output, &output_size); + TEST_CHECK(ret != 0); + TEST_CHECK(output == NULL && output_size == 0); + flb_free(storage); +} + static void test_invalid_input(void) { void *output; @@ -281,6 +355,7 @@ TEST_LIST = { {"advertised_sizes", test_advertised_sizes}, {"invalid_input", test_invalid_input}, {"snappy_framed_checksum", test_snappy_framed_checksum}, + {"snappy_legacy_framed", test_snappy_legacy_framed}, {"context_output_growth", test_context_output_growth}, {"zstd_window_budget", test_zstd_window_budget}, {"gzip_optional_header", test_gzip_optional_header}, From f59aab9aa5798f312a471d687fc99d263d3a999c Mon Sep 17 00:00:00 2001 From: Hiroshi Hatake Date: Tue, 20 Oct 2026 16:23:37 +0900 Subject: [PATCH 4/4] tests: integration: cover Snappy HTTP content encoding Signed-off-by: Hiroshi Hatake --- .../tests/test_in_opentelemetry_001.py | 59 ++++++++++++++++++- 1 file changed, 58 insertions(+), 1 deletion(-) diff --git a/tests/integration/scenarios/in_opentelemetry/tests/test_in_opentelemetry_001.py b/tests/integration/scenarios/in_opentelemetry/tests/test_in_opentelemetry_001.py index ef374cb424f..8fdbcb509ba 100644 --- a/tests/integration/scenarios/in_opentelemetry/tests/test_in_opentelemetry_001.py +++ b/tests/integration/scenarios/in_opentelemetry/tests/test_in_opentelemetry_001.py @@ -37,7 +37,7 @@ # local imports from utils.data_utils import read_json_file from utils.fluent_bit_manager import FluentBitStartupError -from utils.http_matrix import PROTOCOL_CASES, run_curl_request +from utils.http_matrix import PROTOCOL_CASES, curl_supports_http2, run_curl_request from utils.oauth2 import duplicate_header_token, track_jwks_requests from utils.input_pause_resume import ( assert_pause_resume_cycles, @@ -2637,3 +2637,60 @@ def test_in_opentelemetry_grpc_preserves_groups_across_chunks(encoding): scope.scope.version, scope.schema_url, scope_id, record.body.string_value) for record in scope.log_records) assert sorted(actual) == sorted(expected) + + +@pytest.mark.parametrize("case", PROTOCOL_CASES[:2], ids=[case["id"] for case in PROTOCOL_CASES[:2]]) +@pytest.mark.parametrize("encoding", ["snappy", "snappy-framed", "snappy-mixed"]) +def test_in_opentelemetry_http_snappy_content_encoding(case, encoding): + if case["http_mode"] == "http2-prior-knowledge" and not curl_supports_http2(): + pytest.skip("Installed curl lacks HTTP/2 support") + service = Service(IN_OPENTELEMETRY_PROTOCOL_CONFIGS[case["config_key"]]) + try: + service.start() + payload = ExportLogsServiceRequest() + scope = payload.resource_logs.add().scope_logs.add() + expected = "snappy HTTP checksum regression " + "x" * 4096 + scope.log_records.add().body.string_value = expected + encoded = encode_grpc_limit_payload(payload.SerializeToString(), encoding) + result = run_curl_request( + f"http://localhost:{service.flb_listener_port}/v1/logs", + encoded, + headers=["Content-Type: application/x-protobuf", "Content-Encoding: snappy"], + http_mode=case["http_mode"], + ) + assert result["status_code"] == 201 + records = service.service.wait_for_condition( + received_grpc_log_records, timeout=20, interval=0.25, + description="Snappy Content-Encoding log delivery", + ) + assert [record.body.string_value for record in records] == [expected] + finally: + service.stop() + + +@pytest.mark.parametrize("frame_type", [0, 1]) +def test_in_opentelemetry_http_snappy_invalid_checksum(frame_type): + service = Service("otlp_http1_cleartext.yaml") + try: + service.start() + payload = service.build_otel_payload("test_logs_001.in.json", "logs") + encoded = payload if frame_type else encode_grpc_limit_payload(payload, "snappy") + checksum = snappy_test_crc32c(payload) ^ 1 + framed = (b"\xff\x06\x00\x00sNaPpY" + bytes([frame_type]) + + (len(encoded) + 4).to_bytes(3, "little") + + checksum.to_bytes(4, "little") + encoded) + url = f"http://localhost:{service.flb_listener_port}/v1/logs" + # Invalid compressed bodies must not produce log records. + for invalid in [framed, framed[:-1], + b"\xff\x06\x00\x00sNaPpY\x01\x04\x00\x00\x00\x00\x00\x00"]: + response = requests.post(url, data=invalid, timeout=5, headers={ + "Content-Type": "application/x-protobuf", "Content-Encoding": "snappy", + }) + assert response.status_code == 400 + assert not received_grpc_log_records() + # The listener must remain usable after malformed compressed requests. + response = service.send_raw_request("/v1/logs", payload) + assert response.status_code == 201 + service.wait_for_signal_count("logs", 1) + finally: + service.stop()