From 535f501d9fcacd675cae6f5a9a29e303fc6effa9 Mon Sep 17 00:00:00 2001 From: mayankpande88 Date: Tue, 7 Apr 2026 00:30:31 +0530 Subject: [PATCH] fix: prevent OOM crash from L7 protocol misidentification MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The ClickHouse eBPF detector matched non-ClickHouse TCP payloads using a 3-byte heuristic (0x01, 0x00, 0x01), causing ch-go to decode garbage varint lengths and attempt 228TB allocations — a fatal OOM that bypasses defer/recover. 3-layer defense: Layer 0 — eBPF detection hardening: Remove non-port-gated is_clickhouse_query() call. ClickHouse native protocol detection now only on ports 9000/8123. Layer 1 — Bounded parsing: Wrap ch-go proto.NewReader with io.LimitReader(payload size). Varint lengths exceeding actual payload now return io.EOF instead of OOM. Layer 2 — Userspace validation: Add structural checks before invoking external libraries: - ClickHouse: validate query code byte + query ID length - Postgres: reject unknown frame types (Q/B/P/C only) - Zookeeper: validate opcode in known range Layer 3 — Protocol reclassification: Track consecutive parse failures per connection. After 3 failures, override the cached protocol to stop further misidentified parsing. Logs a warning for observability. --- containers/container.go | 47 ++++++++++++++++++++++++++++++++++++- ebpftracer/ebpf/l7/l7.c | 2 -- ebpftracer/l7/clickhouse.go | 26 +++++++++++++++++++- ebpftracer/l7/postgres.go | 9 +++++++ ebpftracer/l7/zookeeper.go | 14 +++++++++++ 5 files changed, 94 insertions(+), 4 deletions(-) diff --git a/containers/container.go b/containers/container.go index af4ea030..d7f26fe3 100644 --- a/containers/container.go +++ b/containers/container.go @@ -102,6 +102,9 @@ type ActiveConnection struct { http2Parser *l7.Http2Parser postgresParser *l7.PostgresParser mysqlParser *l7.MysqlParser + + parseFailCount int + protocolOverride l7.Protocol // non-zero = override eBPF-detected protocol } type ListenDetails struct { @@ -1088,8 +1091,16 @@ func (c *Container) onL7RequestWithResult(pid uint32, fd uint64, timestamp uint6 trace = c.tracer.NewTrace(conn.DestinationKey.ActualDestinationIfKnown(), conn.srcWorkload, destWorkload, actualDestWorkload) } + // Protocol reclassification: if previous parse attempts failed repeatedly, + // the connection was likely misidentified by eBPF heuristics. Use the + // override to skip further parsing for this connection. + protocol := r.Protocol + if conn.protocolOverride != 0 { + protocol = conn.protocolOverride + } + // Process L7 requests and update metrics - switch r.Protocol { + switch protocol { case l7.ProtocolDNS: status := r.Status.DNS() if status == "" { @@ -1282,6 +1293,11 @@ func (c *Container) onL7RequestWithResult(pid uint32, fd uint64, timestamp uint6 conn.postgresParser = l7.NewPostgresParser() } query := conn.postgresParser.Parse(r.Payload) + if query == "" && r.Method != l7.MethodStatementClose { + c.trackParseFail(conn, pid, fd, r.Protocol) + } else { + conn.parseFailCount = 0 + } if trace != nil { trace.PostgresQuery(query, r.Status.Error(), r.Duration) } @@ -1331,6 +1347,11 @@ func (c *Container) onL7RequestWithResult(pid uint32, fd uint64, timestamp uint6 // Update stats for Clickhouse c.l7Stats.observe(r.Protocol, r.Status.String(), "", "", r.Duration, conn.DestinationKey, conn.srcWorkload, r, "") query := l7.ParseClickhouse(r.Payload) + if query == "" { + c.trackParseFail(conn, pid, fd, r.Protocol) + } else { + conn.parseFailCount = 0 + } if trace != nil { trace.ClickhouseQuery(query, r.Status.Error(), r.Duration) } @@ -1338,6 +1359,11 @@ func (c *Container) onL7RequestWithResult(pid uint32, fd uint64, timestamp uint6 // Update stats for Zookeeper c.l7Stats.observe(r.Protocol, r.Status.Zookeeper(), "", "", r.Duration, conn.DestinationKey, conn.srcWorkload, r, "") op, arg := l7.ParseZookeeper(r.Payload) + if op == "" { + c.trackParseFail(conn, pid, fd, r.Protocol) + } else { + conn.parseFailCount = 0 + } if trace != nil { trace.ZookeeperRequest(op, arg, r.Status, r.Duration) } @@ -1351,6 +1377,25 @@ func (c *Container) onL7RequestWithResult(pid uint32, fd uint64, timestamp uint6 return nil, L7RequestProcessed } +const ( + parseFailThreshold = 3 + protocolReclassified = l7.Protocol(0xFF) // sentinel: no protocol matches, hits default case +) + +// trackParseFail tracks consecutive parse failures on a connection. After +// parseFailThreshold failures, the connection's protocol is overridden to +// skip further parsing. This handles eBPF protocol misidentification where +// weak heuristics (e.g., 3-byte ClickHouse check) tag a non-matching +// connection permanently. +func (c *Container) trackParseFail(conn *ActiveConnection, pid uint32, fd uint64, proto l7.Protocol) { + conn.parseFailCount++ + if conn.parseFailCount == parseFailThreshold { + conn.protocolOverride = protocolReclassified + klog.Warningf("reclassified connection pid=%d fd=%d from %s to unknown after %d consecutive parse failures", + pid, fd, proto, conn.parseFailCount) + } +} + // processHTTP2WithoutConnection handles HTTP/2 events when TCP connection tracking failed. // This is common for Go TLS connections where goroutines switch threads between // TCP connect and TLS write, causing fd_by_pid_tgid lookup to fail in eBPF. diff --git a/ebpftracer/ebpf/l7/l7.c b/ebpftracer/ebpf/l7/l7.c index c6a2169b..9c473c47 100644 --- a/ebpftracer/ebpf/l7/l7.c +++ b/ebpftracer/ebpf/l7/l7.c @@ -408,8 +408,6 @@ int trace_enter_write(void *ctx, __u64 fd, __u16 is_tls, char *buf, __u64 size, req->protocol = PROTOCOL_RABBITMQ; } else if (is_amqp_method_frame(payload, size)) { req->protocol = PROTOCOL_RABBITMQ; - } else if (is_clickhouse_query(payload, size)) { - req->protocol = PROTOCOL_CLICKHOUSE; } else if (is_zk_request(payload, total_size)) { req->protocol = PROTOCOL_ZOOKEEPER; } else if (is_kafka_request(payload, size, &req->request_id)) { diff --git a/ebpftracer/l7/clickhouse.go b/ebpftracer/l7/clickhouse.go index a31a25fc..626efbac 100644 --- a/ebpftracer/l7/clickhouse.go +++ b/ebpftracer/l7/clickhouse.go @@ -2,10 +2,13 @@ package l7 import ( "bytes" + "io" "github.com/ClickHouse/ch-go/proto" ) +const clickhouseClientCodeQuery = 1 + func ParseClickhouse(payload []byte) (result string) { // Recover from panics caused by malformed/incomplete packets defer func() { @@ -14,7 +17,28 @@ func ParseClickhouse(payload []byte) (result string) { } }() - r := proto.NewReader(bytes.NewReader(payload)) + // Layer 2: Structural validation before invoking ch-go. + // Reject misidentified payloads early, before the library can + // decode garbage varint lengths and attempt unbounded allocations. + if len(payload) < 3 { + return "" + } + if payload[0] != clickhouseClientCodeQuery { + return "" + } + // Query ID length is varint-encoded. Single-byte varints (0-127) cover + // all practical query IDs. Multi-byte varints (>127) in this position + // indicate garbage data from a misidentified connection. + if payload[1] > 127 { + return "" + } + + // Layer 1: Bound the reader to the actual payload size. + // ch-go's proto.Reader decodes varint string lengths and pre-allocates + // that many bytes. With corrupted data, the varint can decode to TB-scale + // values, causing a fatal OOM that bypasses defer/recover. + // LimitReader caps reads at len(payload), turning the OOM into io.EOF. + r := proto.NewReader(io.LimitReader(bytes.NewReader(payload), int64(len(payload)))) var err error if _, err = r.Byte(); err != nil { return "" diff --git a/ebpftracer/l7/postgres.go b/ebpftracer/l7/postgres.go index 148a120c..5d23c254 100644 --- a/ebpftracer/l7/postgres.go +++ b/ebpftracer/l7/postgres.go @@ -26,6 +26,15 @@ func (p *PostgresParser) Parse(payload []byte) string { return "" } cmd := payload[0] + // Reject unknown frame types early — Postgres detection in eBPF is weak + // (single-byte check), so misidentified connections can reach here. + // Accept all valid frontend message types per the Postgres wire protocol. + switch cmd { + case PostgresFrameQuery, PostgresFrameBind, PostgresFrameParse, PostgresFrameClose, + 'E', 'D', 'S', 'H', 'X': // Execute, Describe, Sync, Flush, Terminate + default: + return "" + } switch cmd { case PostgresFrameQuery: var query string diff --git a/ebpftracer/l7/zookeeper.go b/ebpftracer/l7/zookeeper.go index 71a04d5d..0f40d5a5 100644 --- a/ebpftracer/l7/zookeeper.go +++ b/ebpftracer/l7/zookeeper.go @@ -108,11 +108,25 @@ func zkReadString(r io.Reader) string { return string(res[:n]) } +func isValidZkOp(op int32) bool { + switch op { + case zkOpCreate, zkOpDelete, zkOpExists, zkOpGetData, zkOpSetData, + zkOpGetAcl, zkOpSetAcl, zkOpGetChildren, zkOpSync, zkOpPing, + zkOpGetChildren2, zkOpCheck, zkOpMulti, zkOpReconfig, + zkOpCreateContainer, zkOpCreateTTL, zkOpClose, zkOpSetAuth, zkOpSetWatches: + return true + } + return false +} + func ParseZookeeper(payload []byte) (string, string) { r := bytes.NewReader(payload) h := zkRequestHeader{} if err := binary.Read(r, binary.BigEndian, &h); err != nil { return "", "" } + if !isValidZkOp(h.OpType) { + return "", "" + } return zkParse(r, h.OpType) }