diff --git a/tests/integration/scenarios/in_http/tests/test_in_http_001.py b/tests/integration/scenarios/in_http/tests/test_in_http_001.py index c278bbc5222..298fd312ba5 100644 --- a/tests/integration/scenarios/in_http/tests/test_in_http_001.py +++ b/tests/integration/scenarios/in_http/tests/test_in_http_001.py @@ -221,6 +221,48 @@ def test_send_data(): raise +@pytest.mark.parametrize("chunk_count", [1, 2]) +def test_snappy_framed_uncompressed_data(chunk_count): + payload = b'{"message":"snappy-framed"}' + chunks = [payload] if chunk_count == 1 else [payload[:12], payload[12:]] + framed = bytearray(b'\xff\x06\x00\x00sNaPpY') + + for chunk in chunks: + # Snappy frames store a masked CRC32C of each uncompressed chunk. + crc = 0xffffffff + for byte in chunk: + crc ^= byte + for _ in range(8): + crc = (crc >> 1) ^ (0x82f63b78 if crc & 1 else 0) + crc = (~crc) & 0xffffffff + checksum = (((crc >> 15) | (crc << 17)) + 0xa282ead8) & 0xffffffff + framed.extend(b'\x01' + (len(chunk) + 4).to_bytes(3, 'little')) + framed.extend(checksum.to_bytes(4, 'little') + chunk) + + service = Service("in_http_config") + try: + service.start() + connection = create_connection('localhost', service.flb_listener_port) + try: + connection.request('POST', '/', body=framed, headers={ + 'Content-Type': 'application/json', + 'Content-Encoding': 'snappy', + }) + response = connection.getresponse() + status = response.status + response.read() + finally: + connection.close() + + forwarded_payloads = service.read_forwarded_payloads() + finally: + service.stop() + + assert status == 201 + assert len(forwarded_payloads) == 1 + assert forwarded_payloads[0][0]['message'] == 'snappy-framed' + + @pytest.mark.parametrize("case", PROTOCOL_CASES, ids=[case["id"] for case in PROTOCOL_CASES]) def test_in_http_protocol_matrix(case): service = Service(IN_HTTP_PROTOCOL_CONFIGS[case["config_key"]]) diff --git a/tests/internal/compression.c b/tests/internal/compression.c index 6ae677fc473..1b7234dfadc 100644 --- a/tests/internal/compression.c +++ b/tests/internal/compression.c @@ -217,6 +217,93 @@ static void test_snappy_legacy_framed(void) TEST_CHECK(output == NULL && output_size == 0); flb_free(storage); } +static void test_snappy_framed_aggregation(void) +{ + unsigned char framed[] = { + FLB_SNAPPY_FRAME_TYPE_STREAM_IDENTIFIER, 6, 0, 0, 's', 'N', 'a', 'P', 'p', 'Y', + FLB_SNAPPY_FRAME_TYPE_UNCOMPRESSED_DATA, 7, 0, 0, + 0x6e, 0x57, 0xf1, 0x21, 'a', 'b', 'c', + FLB_SNAPPY_FRAME_TYPE_UNCOMPRESSED_DATA, 7, 0, 0, + 0x6e, 0x57, 0xf1, 0x21, 'a', 'b', 'c' + }; + unsigned char mixed_framed[] = { + FLB_SNAPPY_FRAME_TYPE_STREAM_IDENTIFIER, 6, 0, 0, 's', 'N', 'a', 'P', 'p', 'Y', + FLB_SNAPPY_FRAME_TYPE_COMPRESSED_DATA, 9, 0, 0, + /* Masked checksum followed by a raw Snappy literal for "abc". */ + 0x6e, 0x57, 0xf1, 0x21, 3, 8, 'a', 'b', 'c', + FLB_SNAPPY_FRAME_TYPE_UNCOMPRESSED_DATA, 7, 0, 0, + 0x6e, 0x57, 0xf1, 0x21, 'a', 'b', 'c' + }; + char *output; + size_t output_size; + int ret; + + /* Omit the second 11-byte chunk for the single-chunk case. */ + ret = flb_snappy_uncompress_framed_data((char *) framed, sizeof(framed) - 11, + &output, &output_size); + TEST_CHECK(ret == 0); + TEST_CHECK(output_size == 3); + TEST_CHECK(output != NULL); + if (output != NULL) { + if (output_size == 3) { + TEST_CHECK(memcmp(output, "abc", 3) == 0); + } + flb_free(output); + } + + ret = flb_snappy_uncompress_framed_data((char *) framed, sizeof(framed), + &output, &output_size); + TEST_CHECK(ret == 0); + TEST_CHECK(output_size == 6); + TEST_CHECK(output != NULL); + if (output != NULL) { + if (output_size == 6) { + TEST_CHECK(memcmp(output, "abcabc", 6) == 0); + } + flb_free(output); + } + + ret = flb_snappy_uncompress_framed_data((char *) mixed_framed, + sizeof(mixed_framed) - 11, + &output, &output_size); + TEST_CHECK(ret == 0); + TEST_CHECK(output_size == 3); + TEST_CHECK(output != NULL); + if (output != NULL) { + if (output_size == 3) { + TEST_CHECK(memcmp(output, "abc", 3) == 0); + } + flb_free(output); + } + + ret = flb_snappy_uncompress_framed_data((char *) mixed_framed, + sizeof(mixed_framed), + &output, &output_size); + TEST_CHECK(ret == 0); + TEST_CHECK(output_size == 6); + TEST_CHECK(output != NULL); + if (output != NULL) { + if (output_size == 6) { + TEST_CHECK(memcmp(output, "abcabc", 6) == 0); + } + flb_free(output); + } + + framed[14] ^= 1; + ret = flb_snappy_uncompress_framed_data((char *) framed, sizeof(framed), + &output, &output_size); + TEST_CHECK(ret != 0); + TEST_CHECK(output == NULL); + TEST_CHECK(output_size == 0); + + framed[14] ^= 1; + framed[25] ^= 1; + ret = flb_snappy_uncompress_framed_data((char *) framed, sizeof(framed), + &output, &output_size); + TEST_CHECK(ret != 0); + TEST_CHECK(output == NULL); + TEST_CHECK(output_size == 0); +} static void test_invalid_input(void) { @@ -356,6 +443,7 @@ TEST_LIST = { {"invalid_input", test_invalid_input}, {"snappy_framed_checksum", test_snappy_framed_checksum}, {"snappy_legacy_framed", test_snappy_legacy_framed}, + {"snappy_framed_aggregation", test_snappy_framed_aggregation}, {"context_output_growth", test_context_output_growth}, {"zstd_window_budget", test_zstd_window_budget}, {"gzip_optional_header", test_gzip_optional_header},