Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
42 changes: 42 additions & 0 deletions tests/integration/scenarios/in_http/tests/test_in_http_001.py
Original file line number Diff line number Diff line change
Expand Up @@ -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"]])
Expand Down
88 changes: 88 additions & 0 deletions tests/internal/compression.c
Original file line number Diff line number Diff line change
Expand Up @@ -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',

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

P1 Badge Decode framed checksums as little-endian before this test

On the big-endian s390x entry in .github/workflows/unit-tests.yaml:277-280 (which runs all internal CTests at lines 324-336), the first new success case cannot pass: flb_snappy_uncompress_framed_data() loads frame_checksum through native-endian uint32_t dereferences at src/flb_snappy.c:223 and src/flb_snappy.c:270, so the wire bytes 6e 57 f1 21 are seen as 0x6e57f121 instead of 0x21f1576e. This makes flb-it-compression fail in that matrix; decode the checksum with an unaligned-safe little-endian read.

Useful? React with 👍 / 👎.

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)
{
Expand Down Expand Up @@ -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},
Expand Down