From c03e50f9ae9f0788117f18bd8af86d4595094c92 Mon Sep 17 00:00:00 2001 From: "timmy.yuan" Date: Thu, 8 Oct 2026 09:38:44 +0800 Subject: [PATCH] fix(cpworker): keep the L2 header stack under zmq slice and count runt drops A zmq output dropped every packet when slice was below 18 bytes, because slice was applied before the fixed 18-byte minimum. The dropped packets were not counted in any output statistic. slice now truncates the payload only. The record always keeps the Ethernet header and the whole VLAN tag stack. A frame whose Ethernet header or tag stack was not fully captured produces no record and is counted as error_drop. The fixed 18-byte minimum is gone: untagged frames need 14 bytes, and QinQ frames keep both tags. --- cpworker/src/output_zmq.c | 64 +++++++----- cpworker/tests/unit/output_zmq.c | 166 +++++++++++++++++++++++++++---- docs/ZMQ-WIRE-FORMAT.md | 22 ++-- 3 files changed, 199 insertions(+), 53 deletions(-) diff --git a/cpworker/src/output_zmq.c b/cpworker/src/output_zmq.c index d5804452..94711c39 100644 --- a/cpworker/src/output_zmq.c +++ b/cpworker/src/output_zmq.c @@ -79,7 +79,8 @@ static void flush_error_info(zmq_output_t *output) if (output->error_info.nb_too_small_packets > 0) { - log_warn("zmq output: nb_too_small_packets=%lu (dropped)", output->error_info.nb_too_small_packets); + log_warn("zmq output: nb_too_small_packets=%lu (dropped: Ethernet header or VLAN tags not fully captured)", + output->error_info.nb_too_small_packets); output->error_info.nb_too_small_packets = 0; } @@ -146,23 +147,51 @@ static inline void zmq_flush_if_stale(zmq_output_t *output, time_t now) } } +// Length of the Ethernet header plus the whole VLAN tag stack, or 0 when the captured +// data ends before that stack does. +static size_t l2_header_len(const uint8_t *pkt_data, uint32_t caplen) +{ + if (caplen < sizeof(struct ether_header)) + return 0; + + size_t offset = sizeof(struct ether_header); + uint16_t ether_type = ntohs(((const struct ether_header *)pkt_data)->ether_type); + while (is_vlan_ethertype(ether_type)) + { + if (offset + sizeof(struct vlan_header) > caplen) + return 0; + const struct vlan_header *vlan_hdr = (const struct vlan_header *)(pkt_data + offset); + ether_type = ntohs(vlan_hdr->ether_type); + offset += sizeof(struct vlan_header); + } + return offset; +} + int zmq_send_packet(output_base_t *self, const struct pcap_pkthdr *header, const uint8_t *pkt_data, int direct) { zmq_output_t *output = (zmq_output_t *)self; - int32_t caplen = header->caplen; - if (output->slice > 0 && output->slice < caplen) - caplen = output->slice; + // The record length is a uint16 that also covers the MPLS header + uint32_t caplen = header->caplen <= 65531 ? header->caplen : 65531; - // Minimum caplen: ethernet header (14) + optional VLAN tag (4) must fit - // to avoid underflow in payload_copy_len calculation - if (caplen < (int32_t)(sizeof(struct ether_header) + sizeof(struct vlan_header))) + // The MPLS header goes after the innermost VLAN tag, so a frame whose Ethernet header + // and tag stack were not fully captured cannot form a record + const size_t l2_len = l2_header_len(pkt_data, caplen); + if (l2_len == 0) { output->error_info.nb_too_small_packets++; + bytes_stats_add(&output->base.stats->error_drop_bytes, header->caplen); + packets_stats_add(&output->base.stats->error_drop_packets, 1); + + zmq_flush_if_stale(output, header->ts.tv_sec); return -1; } - uint16_t length = (uint16_t)(caplen <= 65531 ? caplen : 65531) + sizeof(mpls_header); + // slice truncates the payload only: the Ethernet header and VLAN tags are always kept + if (output->slice > 0 && (uint32_t)output->slice < caplen) + caplen = (uint32_t)output->slice > l2_len ? (uint32_t)output->slice : l2_len; + + uint16_t length = (uint16_t)caplen + sizeof(mpls_header); zmq_pkts_buf_t *pkts_buf = &output->pkts_buf; if (direct == PKT_DIR_UNKNOWN) @@ -223,22 +252,7 @@ int zmq_send_packet(output_base_t *self, const struct pcap_pkthdr *header, const struct ether_header eth_hdr_copy; memcpy(ð_hdr_copy, pkt_data, sizeof(struct ether_header)); - // Calculate total VLAN header size by looping through stacked VLANs. - // `length` includes the MPLS header; only `data_len` bytes of pkt_data were captured. - const size_t data_len = length - sizeof(mpls_header); - uint16_t ether_type = ntohs(eth_hdr_copy.ether_type); - size_t vlan_total_size = 0; - - while (is_vlan_ethertype(ether_type)) - { - size_t vlan_offset = sizeof(struct ether_header) + vlan_total_size; - if (vlan_offset + sizeof(struct vlan_header) > data_len) - break; - struct vlan_header *vlan_hdr = (struct vlan_header *)(pkt_data + vlan_offset); - ether_type = ntohs(vlan_hdr->ether_type); - vlan_total_size += sizeof(struct vlan_header); - } - + const size_t vlan_total_size = l2_len - sizeof(struct ether_header); const bool has_vlan = (vlan_total_size > 0); // Write Ethernet header (keep original EtherType if has VLAN, else set MPLS) @@ -266,7 +280,7 @@ int zmq_send_packet(output_base_t *self, const struct pcap_pkthdr *header, const // Copy payload (everything after Ethernet + all VLANs) const size_t payload_offset = sizeof(struct ether_header) + vlan_total_size; - const size_t payload_copy_len = data_len - payload_offset; + const size_t payload_copy_len = caplen - payload_offset; memcpy(&(pkts_buf->buf[buff_pos]), pkt_data + payload_offset, payload_copy_len); buff_pos += payload_copy_len; diff --git a/cpworker/tests/unit/output_zmq.c b/cpworker/tests/unit/output_zmq.c index 21c745ec..f537df97 100644 --- a/cpworker/tests/unit/output_zmq.c +++ b/cpworker/tests/unit/output_zmq.c @@ -145,17 +145,18 @@ void test_destroy_empty_batch_sends_nothing(void) TEST_ASSERT_EQUAL_INT(-1, recv_batch()); } -/* ---- #231: VLAN walk must stay within the captured data ---- */ +/* ---- #231/#271: VLAN walk and slice keep the whole L2 header stack ---- */ // Sends one frame, flushes, and checks the record: Ethernet + VLAN tags (innermost -// EtherType rewritten to MPLS) + MPLS header + remaining payload, all within caplen. -static void assert_vlan_record(const uint8_t *frame, uint32_t caplen, int slice, size_t expected_vlan_size) +// EtherType rewritten to MPLS) + MPLS header + remaining payload. `data_len` is the +// number of captured bytes the record must carry. +static void assert_vlan_record(const uint8_t *frame, uint32_t caplen, int slice, uint32_t data_len, + size_t expected_vlan_size) { char errbuf[ERROR_BUFFER_SIZE] = {0}; zmq_output_t *output = new_output(slice, (char *)VALID_UUID, errbuf); TEST_ASSERT_NOT_NULL_MESSAGE(output, errbuf); - const uint32_t data_len = (slice > 0 && (uint32_t)slice < caplen) ? (uint32_t)slice : caplen; TEST_ASSERT_EQUAL_INT(0, send_frame(output, frame, caplen)); TEST_ASSERT_EQUAL_UINT32(BATCH_HDR_SIZE + RECORD_HDR_SIZE + data_len + MPLS_HDR_SIZE, output->pkts_buf.batch_bufpos); @@ -176,31 +177,120 @@ static void assert_vlan_record(const uint8_t *frame, uint32_t caplen, int slice, const size_t payload_len = data_len - l2_size; if (payload_len > 0) TEST_ASSERT_EQUAL_UINT8_ARRAY(frame + l2_size, rec + l2_size + MPLS_HDR_SIZE, payload_len); + + TEST_ASSERT_EQUAL_UINT64(1, stats.fwd_packets.packets); + TEST_ASSERT_EQUAL_UINT64(0, stats.error_drop_packets.packets); +} + +// Sends one frame that cannot form a record and checks that it is counted as an error drop. +static void assert_error_drop(const uint8_t *frame, uint32_t caplen, int slice) +{ + char errbuf[ERROR_BUFFER_SIZE] = {0}; + zmq_output_t *output = new_output(slice, (char *)VALID_UUID, errbuf); + TEST_ASSERT_NOT_NULL_MESSAGE(output, errbuf); + + TEST_ASSERT_EQUAL_INT(-1, send_frame(output, frame, caplen)); + TEST_ASSERT_EQUAL_UINT16(0, output->pkts_buf.batch_hdr.pkts_num); + TEST_ASSERT_EQUAL_UINT64(1, stats.error_drop_packets.packets); + TEST_ASSERT_EQUAL_UINT64(caplen, stats.error_drop_bytes.bytes); + + zmq_output_destroy(&output->base); + TEST_ASSERT_EQUAL_INT(-1, recv_batch()); + TEST_ASSERT_EQUAL_UINT64(0, stats.fwd_packets.packets); +} + +// Writes an 802.1Q tag at `offset` (TCI, then the EtherType that follows it). +static void put_tag(uint8_t *frame, size_t offset, uint16_t vid, uint16_t next_type) +{ + frame[offset] = (uint8_t)(vid >> 8); + frame[offset + 1] = (uint8_t)vid; + frame[offset + 2] = (uint8_t)(next_type >> 8); + frame[offset + 3] = (uint8_t)next_type; } -void test_vlan_stack_cut_by_slice(void) +void test_slice_keeps_whole_vlan_stack(void) { - // Reproducer from #231: 0x9200 -> 0x8100 -> 0x9100 -> 0x9200 -> 0x86dd, slice cuts the stack + // Reproducer from #231: 0x9200 -> 0x8100 -> 0x9100 -> 0x9200 -> 0x86dd static uint8_t frame[1500]; memset(frame, 0, sizeof(frame)); frame[12] = 0x92; frame[13] = 0x00; - frame[16] = 0x81; - frame[17] = 0x00; - frame[20] = 0x91; - frame[21] = 0x00; - frame[24] = 0x92; - frame[25] = 0x00; - frame[28] = 0x86; - frame[29] = 0xdd; + put_tag(frame, 14, 1, 0x8100); + put_tag(frame, 18, 2, 0x9100); + put_tag(frame, 22, 3, 0x9200); + put_tag(frame, 26, 4, 0x86dd); + + // slice 26 would cut the 4th tag: the record keeps Ethernet + 4 tags = 30 bytes + assert_vlan_record(frame, sizeof(frame), 26, 30, 16); +} + +void test_slice_below_ethernet_and_vlan_keeps_one_tag(void) +{ + // #271 comment: slice 10 used to drop every packet + uint8_t frame[100]; + memset(frame, 0x5a, sizeof(frame)); + frame[12] = 0x81; + frame[13] = 0x00; + put_tag(frame, 14, 100, 0x0800); + + assert_vlan_record(frame, sizeof(frame), 10, 18, 4); +} + +void test_slice_keeps_both_qinq_tags(void) +{ + // 802.1ad outer + 802.1Q inner: slice 18 would cut the inner tag + uint8_t frame[100]; + memset(frame, 0x5a, sizeof(frame)); + frame[12] = 0x88; + frame[13] = 0xa8; + put_tag(frame, 14, 100, 0x8100); + put_tag(frame, 18, 200, 0x0800); + + assert_vlan_record(frame, sizeof(frame), 18, 22, 8); +} + +void test_slice_below_ethernet_keeps_untagged_header(void) +{ + uint8_t frame[100]; + memset(frame, 0x5a, sizeof(frame)); + frame[12] = 0x08; + frame[13] = 0x00; - // 26 captured bytes hold the Ethernet header and 3 complete tags - assert_vlan_record(frame, sizeof(frame), 26, 12); + assert_vlan_record(frame, sizeof(frame), 4, 14, 0); } -void test_vlan_only_frame_without_slice(void) +void test_untagged_14_byte_frame_is_sent(void) { - // A minimum-size frame filled with 0x8100 tags: no slice needed to overrun the walk + uint8_t frame[14]; + memset(frame, 0x5a, sizeof(frame)); + frame[12] = 0x08; + frame[13] = 0x00; + + assert_vlan_record(frame, sizeof(frame), 0, 14, 0); +} + +void test_frame_shorter_than_ethernet_is_error_drop(void) +{ + uint8_t frame[13]; + memset(frame, 0x5a, sizeof(frame)); + + assert_error_drop(frame, sizeof(frame), 0); +} + +void test_frame_with_cut_vlan_tag_is_error_drop(void) +{ + // 16 bytes: Ethernet header with 0x8100, then only half of the tag + uint8_t frame[16]; + memset(frame, 0x5a, sizeof(frame)); + frame[12] = 0x81; + frame[13] = 0x00; + + assert_error_drop(frame, sizeof(frame), 0); +} + +void test_vlan_stack_running_past_caplen_is_error_drop(void) +{ + // A minimum-size frame filled with 0x8100 tags: the stack never ends within caplen uint8_t frame[60]; memset(frame, 0xaa, 12); for (size_t i = 12; i + 1 < sizeof(frame); i += 4) @@ -211,8 +301,33 @@ void test_vlan_only_frame_without_slice(void) frame[i + 3] = 0x01; } - // 60 captured bytes hold the Ethernet header and 11 complete tags, 2 bytes left - assert_vlan_record(frame, sizeof(frame), 0, 44); + assert_error_drop(frame, sizeof(frame), 0); +} + +void test_runt_and_normal_frame_both_accounted(void) +{ + // #271: a runt followed by a normal frame must give fwd + error_drop == 2 + char errbuf[ERROR_BUFFER_SIZE] = {0}; + zmq_output_t *output = new_output(0, (char *)VALID_UUID, errbuf); + TEST_ASSERT_NOT_NULL_MESSAGE(output, errbuf); + + uint8_t runt[16]; + memset(runt, 0x5a, sizeof(runt)); + runt[12] = 0x81; + runt[13] = 0x00; + uint8_t frame[64]; + memset(frame, 0x5a, sizeof(frame)); + frame[12] = 0x08; + frame[13] = 0x00; + + TEST_ASSERT_EQUAL_INT(-1, send_frame(output, runt, sizeof(runt))); + TEST_ASSERT_EQUAL_INT(0, send_frame(output, frame, sizeof(frame))); + zmq_output_destroy(&output->base); + + TEST_ASSERT_GREATER_THAN_INT(0, recv_batch()); + TEST_ASSERT_EQUAL_UINT16(1, batch_pkts_num()); + TEST_ASSERT_EQUAL_UINT64(1, stats.fwd_packets.packets); + TEST_ASSERT_EQUAL_UINT64(1, stats.error_drop_packets.packets); } int main(void) @@ -225,8 +340,15 @@ int main(void) RUN_TEST(test_destroy_flushes_pending_batch); RUN_TEST(test_destroy_empty_batch_sends_nothing); - RUN_TEST(test_vlan_stack_cut_by_slice); - RUN_TEST(test_vlan_only_frame_without_slice); + RUN_TEST(test_slice_keeps_whole_vlan_stack); + RUN_TEST(test_slice_below_ethernet_and_vlan_keeps_one_tag); + RUN_TEST(test_slice_keeps_both_qinq_tags); + RUN_TEST(test_slice_below_ethernet_keeps_untagged_header); + RUN_TEST(test_untagged_14_byte_frame_is_sent); + RUN_TEST(test_frame_shorter_than_ethernet_is_error_drop); + RUN_TEST(test_frame_with_cut_vlan_tag_is_error_drop); + RUN_TEST(test_vlan_stack_running_past_caplen_is_error_drop); + RUN_TEST(test_runt_and_normal_frame_both_accounted); return UNITY_END(); } diff --git a/docs/ZMQ-WIRE-FORMAT.md b/docs/ZMQ-WIRE-FORMAT.md index 1e0880f1..800ed501 100644 --- a/docs/ZMQ-WIRE-FORMAT.md +++ b/docs/ZMQ-WIRE-FORMAT.md @@ -42,9 +42,10 @@ Configuration (`cpworker/examples/libpcap_zmq.json`): } ``` -> `slice` (output-level, not shown above) truncates each packet to at most `slice` bytes. -> Set it to `0`/unset when the full payload is needed. Note that `slice` still leaves at -> least an Ethernet + one VLAN header so the record framing stays valid. +> `slice` (output-level, not shown above) truncates the payload of each packet to at most +> `slice` bytes. Set it to `0`/unset when the full payload is needed. `slice` never cuts into +> the Ethernet header or the VLAN tag stack: a `slice` smaller than those headers keeps +> exactly the headers (§4.1). --- @@ -95,11 +96,20 @@ a heartbeat by inspecting the EtherType of the frame (§7). In `zmq_send_packet()`: -```c -uint16_t length = (uint16_t)(caplen <= 65531 ? caplen : 65531) + sizeof(mpls_header); ``` +caplen = min(captured_length, 65531) +l2_len = 14 + 4 * (number of VLAN tags) // Ethernet header + whole tag stack +if slice > 0 and slice < caplen: + caplen = max(slice, l2_len) +length = caplen + 4 // + MPLS-like tag +``` + +and `pkt_data_len == length`. VLAN tags are recognised by the EtherTypes `0x8100`, `0x88a8`, +`0x9100` and `0x9200`. -i.e. `length = min(captured_length, 65531) + 4`, and `pkt_data_len == length`. +A frame whose Ethernet header or tag stack is not fully captured (fewer than 14 bytes, or a +tag EtherType with no complete tag after it) produces no record. It is counted in the +output's `error_drop_packets` / `error_drop_bytes` statistics. ---