Truncated packets

[buger]

Hi! Any reason why you have fixed max packet payload size to 1500? I have MTUs which is bigger then this, like 65536, and it for sure gets cut. Do you think it can be set automatically based on MTU? I guess it mostly affects memory usage?

[mozillazg]

Any reason why you have fixed max packet payload size to 1500?

Because it's the recommended value for most interfaces and the -s flag is not implemented yet.

Do you think it can be set automatically based on MTU?

I'll try to implement the -s/--snapshot-length flag like tcpdump, after that, this issue will be fixed.

[mozillazg]

@buger Please try version of #78, it implemented the -s/--snapshot-length flag and allowed to change the max packet payload size (default 262144, same as tcpdump) : https://github.com/mozillazg/ptcpdump/actions/runs/9647201633

[buger]

Thanks! I also tried to play with it by myself, to implement similar functionality without limit (well almost).

#define MAX_PACKET_SIZE 65535
#define MIN(a, b) ((a) < (b) ? (a) : (b))

static __always_inline int process_packet_chunk(struct __sk_buff *skb, u32 packet_id, u32 *offset, u16 *chunk_index, bool egress, u32 packet_size) {
    struct packet_event_t *event;
    u32 chunk_size;

    TRACE("Processing chunk: packet_id=%u, offset=%u, packet_size=%u",
               packet_id, *offset, packet_size);

    // Validate offset
    if (*offset >= packet_size) {
        TRACE("Invalid offset: packet_id=%u, offset=%u, packet_size=%u",
                   packet_id, *offset, packet_size);
        return -1;
    }

    // Calculate and validate chunk_size
    chunk_size = MIN(packet_size - *offset, MAX_PAYLOAD_SIZE);
    if (chunk_size == 0) {
        TRACE("Zero chunk size: packet_id=%u, offset=%u, packet_size=%u",
                   packet_id, *offset, packet_size);
        return -1;
    }

    // Additional check for verifier
    if (chunk_size > MAX_PAYLOAD_SIZE) {
        bpf_printk("Chunk size too large: packet_id=%u, offset=%u, chunk_size=%u",
                   packet_id, *offset, chunk_size);
        return -1;
    }

    TRACE("Chunk size calculated: packet_id=%u, offset=%u, chunk_size=%u",
               packet_id, *offset, chunk_size);

    event = bpf_ringbuf_reserve(&packet_flow, sizeof(struct packet_event_t), 0);
    if (!event) {
        TRACE("Failed to reserve ringbuf: packet_id=%u, offset=%u, chunk_size=%u",
                   packet_id, *offset, chunk_size);
        return -1;
    }

    __builtin_memset(&event->meta, 0, sizeof(event->meta));

    if (chunk_size < 2) {
        chunk_size = 2;
    }

    if (*offset + chunk_size > packet_size) {
        chunk_size = 1;
    }

    // Use bpf_skb_load_bytes to read packet data
    if (bpf_skb_load_bytes(skb, *offset, event->payload, chunk_size) < 0) {
        TRACE("Failed to load packet data: packet_id=%u, offset=%u, chunk_size=%u",
                   packet_id, *offset, chunk_size);
        bpf_ringbuf_discard(event, 0);
        return -1;
    }

    event->meta.packet_id = packet_id;
    event->meta.packet_type = egress ? EGRESS_PACKET : INGRESS_PACKET;
    event->meta.timestamp = bpf_ktime_get_ns();
    event->meta.ifindex = skb->ifindex;
    event->meta.packet_size = packet_size;
    event->meta.payload_len = chunk_size;
    event->meta.chunk_index = *chunk_index;
    event->meta.is_last_chunk = (*offset + chunk_size) >= packet_size ? 1 : 0;

    // if (pid_meta.pid > 0) {
    //     event->meta.process.pid = pid_meta.pid;
    //     event->meta.process.mntns_id = pid_meta.mntns_id;
    //     event->meta.process.netns_id = pid_meta.netns_id;
    //     __builtin_memcpy(&event->meta.process.cgroup_name, &pid_meta.cgroup_name, sizeof(pid_meta.cgroup_name));
    // }

    bpf_ringbuf_submit(event, 0);

    *offset += chunk_size;
    (*chunk_index)++;

    TRACE("Successfully processed chunk: packet_id=%u, offset=%u, chunk_size=%u, chunk_index=%u",
               packet_id, *offset, chunk_size, *chunk_index);

    return 0;
}

static __always_inline void handle_tc(struct __sk_buff *skb, bool egress) {
    // Ensure we can access the packet data
    if (bpf_skb_pull_data(skb, 0) < 0) {
        // bpf_printk("Failed to pull skb data in handle_tc");
        return;
    }

    u32 packet_size = skb->len;
    long skb_size = (void *)(long)skb->data_end - (void *)(long)skb->data;

    if (!pcap_filter((void *)(long)skb->data, (void *)(long)skb->data_end, (void *)skb, (void *)(long)skb->data, (void *)(long)skb->data_end)) {
        // bpf_printk("Packet filtered out by pcap_filter");
        return;
    }

    if (packet_size <= skb_size) {
        // We need packets only with data in it
        // bpf_printk("Packet size is less than packet_size: packet_size=%u, skb_size=%ld", packet_size, skb_size);
        return;
    }

    // struct process_meta_t pid_meta = {0};
    // if (get_pid_meta(skb, &pid_meta, egress) < 0) {
    //     TRACE("Failed to get pid meta");
    //     return;
    // }

    u32 offset = 0;
    u16 chunk_index = 0;
    u32 packet_id = bpf_get_prandom_u32();

    TRACE("Starting packet processing: packet_id=%u, packet_size=%u, skb packet size: %u", packet_id, packet_size, skb_size);

    #pragma unroll
    for (int i = 0; i < 4; i++) {
        if (offset >= packet_size) {
            TRACE("Reached end of packet: packet_id=%u, offset=%u, packet_size=%u",
                       packet_id, offset, packet_size);
            break;
        }
        if (process_packet_chunk(skb, packet_id, &offset, &chunk_index, egress, packet_size) < 0) {
            TRACE("Failed to process chunk: packet_id=%u, chunk_index=%u", packet_id, chunk_index);
            break;
        }
    }

    TRACE("Finished packet processing: packet_id=%u, final_offset=%u, final_chunk_index=%u",
               packet_id, offset, chunk_index);
}

As you can see I switch to ring buff here, because were unable to make it work current way, and after implemented packet reconcilation logic on app level:

type BpfPacketEvent struct {
    BpfPacketEventT

    FullPayload []byte
}

type packetChunk struct {
    event      BpfPacketEventT
    receivedAt time.Time
}

type packetAssembler struct {
    chunks          map[uint32][]packetChunk
    mutex           sync.Mutex
    cleanupInterval time.Duration
}

func newPacketAssembler() *packetAssembler {
    pa := &packetAssembler{
        chunks:          make(map[uint32][]packetChunk),
        cleanupInterval: 30 * time.Second,
    }
    go pa.periodicCleanup()
    return pa
}

func (pa *packetAssembler) addChunk(event BpfPacketEventT) (*BpfPacketEvent, bool) {
    pa.mutex.Lock()
    defer pa.mutex.Unlock()

    chunk := packetChunk{
        event:      event,
        receivedAt: time.Now(),
    }

    packetID := event.Meta.PacketId
    pa.chunks[packetID] = append(pa.chunks[packetID], chunk)

    if event.Meta.IsLastChunk && event.Meta.ChunkIndex == uint8(len(pa.chunks[packetID])-1) {
        completeEvent := pa.assemblePacket(packetID)
        delete(pa.chunks, packetID)

        if event.Meta.PacketSize > 6000 {
            log.Warn().Msgf("at the moment EBPF input supports MTU only up to 6000, and your single packet size is bigger than it; actual size: %d", event.Meta.PacketSize)
        }
        return completeEvent, true
    }

    return nil, false
}

func (pa *packetAssembler) assemblePacket(packetID uint32) *BpfPacketEvent {
    chunks := pa.chunks[packetID]

    if len(chunks) == 1 {
        // If there's only one chunk, reuse its payload without copying
        return &BpfPacketEvent{
            BpfPacketEventT: chunks[0].event,
            FullPayload:     chunks[0].event.Payload[:chunks[0].event.Meta.PayloadLen],
        }
    }

    totalSize := 0
    for _, chunk := range chunks {
        totalSize += int(chunk.event.Meta.PayloadLen)
    }

    completeEvent := BpfPacketEvent{
        BpfPacketEventT: chunks[0].event, // Preserve metadata from the first chunk
        FullPayload:     make([]byte, totalSize),
    }

    offset := 0
    for _, chunk := range chunks {
        copy(completeEvent.FullPayload[offset:], chunk.event.Payload[:chunk.event.Meta.PayloadLen])
        offset += int(chunk.event.Meta.PayloadLen)
    }

    return &completeEvent
}

func (pa *packetAssembler) periodicCleanup() {
    ticker := time.NewTicker(pa.cleanupInterval)
    defer ticker.Stop()

    for range ticker.C {
        pa.cleanup()
    }
}

func (pa *packetAssembler) cleanup() {
    pa.mutex.Lock()
    defer pa.mutex.Unlock()

    now := time.Now()
    for packetID, chunks := range pa.chunks {
        if now.Sub(chunks[0].receivedAt) > pa.cleanupInterval {
            delete(pa.chunks, packetID)
        }
    }
}

func (b *BPF) PullPacketEvents(ctx context.Context, chanSize int) (<-chan BpfPacketEvent, error) {
    // reader, err := perf.NewReader(b.objs.PacketEvents, 1500*1000)
    // if err != nil {
    //  return nil, xerrors.Errorf(": %w", err)
    // }
    ch := make(chan BpfPacketEvent, chanSize)
    go func() {
        defer close(ch)
        defer b.Close()
        // defer reader.Close()
        b.handlePacketEvents(ctx, ch)
    }()

    return ch, nil
}