Skip to content

fix(cpworker): keep the L2 header stack under zmq slice and count runt drops #302

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
timmy21 merged 1 commit into from
Oct 8, 2026
Merged

fix(cpworker): keep the L2 header stack under zmq slice and count runt drops #302

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
64 changes: 39 additions & 25 deletions cpworker/src/output_zmq.c
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
}

Expand Down Expand Up @@ -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)
Expand Down Expand Up @@ -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(&eth_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)
Expand Down Expand Up @@ -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;

Expand Down
166 changes: 144 additions & 22 deletions cpworker/tests/unit/output_zmq.c
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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);
Expand All @@ -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)
Expand All @@ -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)
Expand All @@ -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();
}
22 changes: 16 additions & 6 deletions docs/ZMQ-WIRE-FORMAT.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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).

---

Expand Down Expand Up @@ -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.

---

Expand Down
Loading