common.go

// Copyright The OpenTelemetry Authors
// SPDX-License-Identifier: Apache-2.0

package ebpfcommon // import "go.opentelemetry.io/obi/pkg/ebpf/common"

import (
	"bufio"
	"bytes"
	"context"
	"errors"
	"fmt"
	"io"
	"log/slog"
	"net"
	"os"
	"strings"
	"sync"
	"time"
	"unsafe"

	"github.com/cilium/ebpf"
	"github.com/cilium/ebpf/asm"
	"github.com/cilium/ebpf/link"
	lru "github.com/hashicorp/golang-lru/v2"
	"github.com/hashicorp/golang-lru/v2/expirable"
	"github.com/hashicorp/golang-lru/v2/simplelru"

	"go.opentelemetry.io/obi/pkg/appolly/app/request"
	"go.opentelemetry.io/obi/pkg/config"
	"go.opentelemetry.io/obi/pkg/ebpf/common/dnsparser"
	ebpfhttp "go.opentelemetry.io/obi/pkg/ebpf/common/http"
	"go.opentelemetry.io/obi/pkg/internal/ebpf/kafkaparser"
	"go.opentelemetry.io/obi/pkg/internal/ebpf/ringbuf"
	"go.opentelemetry.io/obi/pkg/internal/largebuf"
	"go.opentelemetry.io/obi/pkg/pipe/msg"
)

//go:generate $BPF2GO -cc $BPF_CLANG -cflags $BPF_CFLAGS -target amd64,arm64 -type http_request_trace_t -type sql_request_trace_t -type http_info_t -type connection_info_t -type http2_grpc_request_t -type tcp_req_t -type kafka_client_req_t -type kafka_go_req_t -type redis_client_req_t -type tcp_large_buffer_t -type otel_span_t -type mongo_go_client_req_t -type dns_req_t Bpf ../../../bpf/common/common.c -- -I../../../bpf

// HTTPRequestTrace contains information from an HTTP request as directly received from the
// eBPF layer. This contains low-level C structures for accurate binary read from ring buffer.
type (
	HTTPRequestTrace     BpfHttpRequestTraceT
	SQLRequestTrace      BpfSqlRequestTraceT
	BPFHTTPInfo          BpfHttpInfoT
	BPFConnInfo          BpfConnectionInfoT
	TCPRequestInfo       BpfTcpReqT
	GoSaramaClientInfo   BpfKafkaClientReqT
	GoRedisClientInfo    BpfRedisClientReqT
	GoKafkaGoClientInfo  BpfKafkaGoReqT
	TCPLargeBufferHeader BpfTcpLargeBufferT
	GoOTelSpanTrace      BpfOtelSpanT
	GoMongoClientInfo    BpfMongoGoClientReqT
	DNSInfo              BpfDnsReqT
)

// Go mirror of tp_info.h -> enum tp_flags
// Values from https://www.w3.org/TR/trace-context/
const (
	TPFlagSampled = 1
)

const (
	EventTypeSQL            = 5  // EVENT_SQL_CLIENT - SQL client event
	EventTypeKHTTP          = 6  // EVENT_K_HTTP_REQUEST - HTTP Events generated by kprobes
	EventTypeKHTTP2         = 7  // EVENT_K_HTTP2_REQUEST - HTTP2/gRPC Events generated by kprobes
	EventTypeTCP            = 8  // EVENT_TCP_REQUEST - Unknown TCP protocol to be classified by user space
	EventTypeGoSarama       = 9  // EVENT_GO_KAFKA - Kafka client for Go (Shopify/IBM Sarama)
	EventTypeGoRedis        = 10 // EVENT_GO_REDIS - Redis client for Go
	EventTypeGoKafkaGo      = 11 // EVENT_GO_KAFKA_SEG - Kafka-Go client from Segment-io
	EventTypeTCPLargeBuffer = 12 // EVENT_TCP_LARGE_BUFFER - Dynamically sized TCP buffers
	EventOTelSDKGo          = 13 // EVENT_GO_SPAN - OTel SDK manual span
	EventTypeGoMongo        = 14 // EVENT_GO_MONGO - Go MongoDB spans
	EventTypeFailedConnect  = 15 // EVENT_FAILED_CONNECT - Failed Connections
	EventTypeDNS            = 16 // EVENT_DNS_REQUEST - DNS events
)

// Kernel-side classification
const (
	ProtocolTypeUnknown uint8 = iota
	ProtocolTypeMySQL
	ProtocolTypePostgres
	ProtocolTypeHTTP // not used, written for consistency
	ProtocolTypeKafka
	ProtocolTypeMQTT // placeholder for future kernel-space detection
	ProtocolTypeMSSQL
	ProtocolTypeNATS // placeholder for future kernel-space detection
	ProtocolTypeAMQP // placeholder for future kernel-space detection
)

const (
	GenericEventSourceTypeKProbes  uint8 = 0
	GenericEventSourceTypeLWThread uint8 = 1
)

var IntegrityModeOverride = false

type TracerCapability uint64

// ProbeDesc holds the information of the instrumentation points of a given
// function/symbol
type ProbeDesc struct {
	// Required, if true, will cancel the execution of the eBPF Tracer
	// if the function has not been found in the executable
	Required bool

	// The eBPF program to attach to the symbol as a uprobe (either to the
	// symbol name or to StartOffset)
	Start *ebpf.Program

	// The eBPF program to attach to the symbol either as a uretprobe or as a
	// uprobe to ReturnOffsets
	End *ebpf.Program

	// Optional offset to the start of the symbol
	StartOffset uint64

	// Optional list of the offsets of every RET instruction in the symbol
	ReturnOffsets []uint64
}

type Filter struct {
	io.Closer
	Fd int
}

type SockOps struct {
	io.Closer
	Program       *ebpf.Program
	AttachAs      ebpf.AttachType
	SockopsCgroup link.Link
}

type SockMsg struct {
	io.Closer
	Program  *ebpf.Program
	MapFD    int
	AttachAs ebpf.AttachType
}

type Iter struct {
	Program *ebpf.Program
	Link    link.Link
}

func (it *Iter) Run(log *slog.Logger) error {
	log.Debug("Running iterator", "iterator", it.Program.String())

	if it.Link == nil {
		return errors.New("iterator link is nil")
	}

	rd, err := it.Link.(*link.Iter).Open()
	if err != nil {
		return fmt.Errorf("open iterator: %w", err)
	}
	defer rd.Close()

	scanner := bufio.NewScanner(rd)
	for scanner.Scan() {
		log.Debug("Iterator output", "line", scanner.Text(), "iterator", it.Program.String())
	}
	if err := scanner.Err(); err != nil {
		return fmt.Errorf("read iterator: %w", err)
	}

	log.Debug("Iterator finished", "iterator", it.Program.String())
	return nil
}

type Tracing struct {
	Program  *ebpf.Program
	AttachAs ebpf.AttachType
	Link     link.Link
}

type MisclassifiedEvent struct {
	EventType int
	TCPInfo   *TCPRequestInfo
}

// CouchbaseBucketInfo holds the bucket, scope, and collection for a Couchbase connection.
type CouchbaseBucketInfo struct {
	Bucket     string
	Scope      string
	Collection string
}

type EBPFParseContext struct {
	protocolDebug              bool
	h2c                        *lru.Cache[uint64, h2Connection]
	redisDBCache               *simplelru.LRU[BpfConnectionInfoT, int]
	couchbaseBucketCache       *simplelru.LRU[BpfConnectionInfoT, CouchbaseBucketInfo]
	largeBuffers               *expirable.LRU[largeBufferKey, *largebuf.LargeBuffer]
	mongoRequestCache          PendingMongoDBRequests
	mysqlPreparedStatements    *simplelru.LRU[mysqlPreparedStatementsKey, string]
	postgresPreparedStatements *simplelru.LRU[postgresPreparedStatementsKey, string]
	postgresPortals            *simplelru.LRU[postgresPortalsKey, string]
	mssqlPreparedStatements    *simplelru.LRU[mssqlPreparedStatementsKey, string]
	kafkaTopicUUIDToName       *simplelru.LRU[kafkaparser.UUID, string]
	payloadExtraction          config.PayloadExtraction
	httpEnricher               *ebpfhttp.HTTPEnricher
	dnsEvents                  *expirable.LRU[dnsparser.DNSId, *request.Span]
	emitSpans                  func([]request.Span)
}

// sharedForwarder is implemented by ringBufForwarder[T] so that
// EBPFEventContext.SharedRingBuffer can hold a type safe reference
// without being generic itself.
// This interface with a simple method is preferred to the any type.
type sharedForwarder interface {
	AlreadyForwarded(ctx context.Context)
}

type EBPFEventContext struct {
	CommonPIDsFilter ServiceFilter
	SharedRingBuffer sharedForwarder
	EBPFMaps         map[string]*ebpf.Map
	RingBufLock      sync.Mutex
	MapsLock         sync.Mutex
	LoadLock         sync.Mutex
	Capabilities     TracerCapability
}

var MisclassifiedEvents = make(chan MisclassifiedEvent)

func ptlog() *slog.Logger { return slog.With("component", "ebpf.ProcessTracer") }

func isASCIIAlnumByte(c byte) bool {
	return (c >= 'a' && c <= 'z') || (c >= 'A' && c <= 'Z') || (c >= '0' && c <= '9')
}

func isASCII(s string) bool {
	for i := 0; i < len(s); i++ {
		c := s[i]
		if isASCIIAlnumByte(c) || c == '.' || c == '_' || c == ' ' || c == '-' {
			continue
		}
		return false
	}

	return true
}

func isASCIIAlnumBytes(field []byte) bool {
	if len(field) == 0 {
		return false
	}

	for _, b := range field {
		if !isASCIIAlnumByte(b) {
			return false
		}
	}

	return true
}

func isASCIIDecimal(field []byte) bool {
	if len(field) == 0 {
		return false
	}

	for _, b := range field {
		if b < '0' || b > '9' {
			return false
		}
	}

	return true
}

func NewEBPFParseContext(cfg *config.EBPFTracer, spansChan *msg.Queue[[]request.Span], filter ServiceFilter) *EBPFParseContext {
	var (
		err                        error
		protocolDebug              bool
		redisDBCache               *simplelru.LRU[BpfConnectionInfoT, int]
		couchbaseBucketCache       *simplelru.LRU[BpfConnectionInfoT, CouchbaseBucketInfo]
		mysqlPreparedStatements    *simplelru.LRU[mysqlPreparedStatementsKey, string]
		postgresPreparedStatements *simplelru.LRU[postgresPreparedStatementsKey, string]
		postgresPortals            *simplelru.LRU[postgresPortalsKey, string]
		mssqlPreparedStatements    *simplelru.LRU[mssqlPreparedStatementsKey, string]
		kafkaTopicUUIDToName       *simplelru.LRU[kafkaparser.UUID, string]
		mongoRequestCache          PendingMongoDBRequests
		payloadExtraction          config.PayloadExtraction
		dnsEvents                  *expirable.LRU[dnsparser.DNSId, *request.Span]
		emitSpans                  func([]request.Span)
	)

	h2c, _ := lru.New[uint64, h2Connection](1024 * 10)
	largeBuffers := expirable.NewLRU[largeBufferKey, *largebuf.LargeBuffer](1024, nil, 5*time.Minute)

	if spansChan != nil {
		emitSpans = func(spans []request.Span) {
			if len(spans) == 0 {
				return
			}
			if filter != nil {
				spans = filter.Filter(spans)
			}
			spansChan.SendCtx(context.Background(), spans)
		}
	}

	if cfg != nil {
		protocolDebug = cfg.ProtocolDebug

		if cfg.RedisDBCache.Enabled {
			redisDBCache, err = simplelru.NewLRU[BpfConnectionInfoT, int](cfg.RedisDBCache.MaxSize, nil)
			if err != nil {
				ptlog().Error("failed to create Redis DB cache", "error", err)
				redisDBCache = nil
			}
		}

		couchbaseBucketCache, err = simplelru.NewLRU[BpfConnectionInfoT, CouchbaseBucketInfo](cfg.CouchbaseDBCacheSize, nil)
		if err != nil {
			ptlog().Error("failed to create Couchbase bucket cache", "error", err)
			couchbaseBucketCache = nil
		}

		mysqlPreparedStatements, err = simplelru.NewLRU[mysqlPreparedStatementsKey, string](cfg.MySQLPreparedStatementsCacheSize, nil)
		if err != nil {
			ptlog().Error("failed to create MySQL prepared statements cache", "error", err)
		}

		postgresPreparedStatements, err = simplelru.NewLRU[postgresPreparedStatementsKey, string](cfg.PostgresPreparedStatementsCacheSize, nil)
		if err != nil {
			ptlog().Error("failed to create Postgres prepared statements cache", "error", err)
		}

		postgresPortals, err = simplelru.NewLRU[postgresPortalsKey, string](cfg.PostgresPreparedStatementsCacheSize, nil)
		if err != nil {
			ptlog().Error("failed to create Postgres portals cache", "error", err)
		}

		mssqlPreparedStatements, err = simplelru.NewLRU[mssqlPreparedStatementsKey, string](cfg.MSSQLPreparedStatementsCacheSize, nil)
		if err != nil {
			ptlog().Error("failed to create MSSQL prepared statements cache", "error", err)
		}

		kafkaTopicUUIDToName, err = simplelru.NewLRU[kafkaparser.UUID, string](cfg.KafkaTopicUUIDCacheSize, nil)
		if err != nil {
			ptlog().Error("failed to create Kafka topic UUID to name cache", "error", err)
		}

		mongoRequestCache = expirable.NewLRU[MongoRequestKey, *MongoRequestValue](cfg.MongoRequestsCacheSize, nil, 0)

		payloadExtraction = cfg.PayloadExtraction

		dnsEvents = expirable.NewLRU(1024, dnsEventExpireHandler(emitSpans), cfg.DNSRequestTimeout)
	}

	var httpEnricher *ebpfhttp.HTTPEnricher
	if payloadExtraction.HTTP.Enrichment.Enabled {
		httpEnricher = ebpfhttp.NewHTTPEnricher(payloadExtraction.HTTP.Enrichment)
	}

	return &EBPFParseContext{
		protocolDebug:              protocolDebug,
		h2c:                        h2c,
		redisDBCache:               redisDBCache,
		couchbaseBucketCache:       couchbaseBucketCache,
		largeBuffers:               largeBuffers,
		mongoRequestCache:          mongoRequestCache,
		mysqlPreparedStatements:    mysqlPreparedStatements,
		postgresPreparedStatements: postgresPreparedStatements,
		postgresPortals:            postgresPortals,
		mssqlPreparedStatements:    mssqlPreparedStatements,
		kafkaTopicUUIDToName:       kafkaTopicUUIDToName,
		payloadExtraction:          payloadExtraction,
		httpEnricher:               httpEnricher,
		dnsEvents:                  dnsEvents,
		emitSpans:                  emitSpans,
	}
}

func (ctx *EBPFParseContext) emitExtraSpans(spans ...request.Span) {
	if ctx == nil || ctx.emitSpans == nil || len(spans) == 0 {
		return
	}

	ctx.emitSpans(spans)
}

func NewEBPFEventContext() *EBPFEventContext {
	return &EBPFEventContext{
		EBPFMaps:    map[string]*ebpf.Map{},
		RingBufLock: sync.Mutex{},
		MapsLock:    sync.Mutex{},
		LoadLock:    sync.Mutex{},
	}
}

func ReadBPFTraceAsSpan(parseCtx *EBPFParseContext, cfg *config.EBPFTracer, record *ringbuf.Record, filter ServiceFilter) (request.Span, bool, error) {
	if len(record.RawSample) == 0 {
		return request.Span{}, true, errors.New("invalid ringbuffer record size")
	}

	eventType := record.RawSample[0]

	switch eventType {
	case EventTypeSQL:
		return ReadSQLRequestTraceAsSpan(record)
	case EventTypeKHTTP:
		return ReadHTTPInfoIntoSpan(parseCtx, record, filter)
	case EventTypeKHTTP2:
		return ReadHTTP2InfoIntoSpan(parseCtx, record, filter)
	case EventTypeTCP:
		return ReadTCPRequestIntoSpan(parseCtx, cfg, record, filter)
	case EventTypeGoSarama:
		return ReadGoSaramaRequestIntoSpan(record)
	case EventTypeGoRedis:
		return ReadGoRedisRequestIntoSpan(record)
	case EventTypeGoMongo:
		return ReadGoMongoRequestIntoSpan(record)
	case EventTypeGoKafkaGo:
		return ReadGoKafkaGoRequestIntoSpan(record)
	case EventTypeTCPLargeBuffer:
		return appendTCPLargeBuffer(parseCtx, record)
	case EventOTelSDKGo:
		return ReadGoOTelEventIntoSpan(record)
	case EventTypeFailedConnect:
		return ReadFailedConnectIntoSpan(record, filter)
	case EventTypeDNS:
		return readDNSEventIntoSpan(parseCtx, record)
	}

	event, err := ReinterpretCast[HTTPRequestTrace](record.RawSample)
	if err != nil {
		return request.Span{}, true, err
	}

	return HTTPRequestTraceToSpan(event), false, nil
}

func ReinterpretCast[T any](b []byte) (*T, error) {
	var zero T

	if len(b) < int(unsafe.Sizeof(zero)) {
		return nil, errors.New("byte slice too short")
	}

	return (*T)(unsafe.Pointer(unsafe.SliceData(b))), nil
}

func ReadSQLRequestTraceAsSpan(record *ringbuf.Record) (request.Span, bool, error) {
	event, err := ReinterpretCast[SQLRequestTrace](record.RawSample)
	if err != nil {
		return request.Span{}, true, err
	}

	return SQLRequestTraceToSpan(event), false, nil
}

type KernelLockdown uint8

const (
	KernelLockdownNone KernelLockdown = iota + 1
	KernelLockdownIntegrity
	KernelLockdownConfidentiality
	KernelLockdownOther
)

func SupportsLogInjection(log *slog.Logger) bool {
	if !hasCapSysAdmin() {
		log.Info("log injection not supported: missing CAP_SYS_ADMIN capability")
		return false
	}

	lockdownMode := KernelLockdownMode()
	if lockdownMode != KernelLockdownNone {
		log.Info("log injection not supported: kernel in lockdown mode")
		return false
	}

	return true
}

func SupportsContextPropagationWithProbe(log *slog.Logger) bool {
	kernelMajor, kernelMinor := KernelVersion()
	log.Debug("Linux kernel version", "major", kernelMajor, "minor", kernelMinor)

	if kernelMajor < 5 || (kernelMajor == 5 && kernelMinor < 10) {
		log.Debug("Found Linux kernel earlier than 5.10, Go trace context propagation at library level is supported", "major", kernelMajor, "minor", kernelMinor)
		return true
	}

	// bpf_probe_write_user(), used to inject the context, requires CAP_SYS_ADMIN

	if !hasCapSysAdmin() {
		log.Info("Go context propagation at library level disabled due to missing capability CAP_SYS_ADMIN")
		return false
	}

	lockdown := KernelLockdownMode()

	if lockdown == KernelLockdownNone {
		log.Debug("Kernel not in lockdown mode, Go trace context propagation at library level is supported.")
		return true
	}

	return false
}

func SupportsEBPFLoops(log *slog.Logger, overrideKernelVersion bool) bool {
	if overrideKernelVersion {
		log.Debug("Skipping kernel version check for bpf_loop functionality: user supplied confirmation of support")
		return true
	}
	kernelMajor, kernelMinor := KernelVersion()
	return kernelMajor > 5 || (kernelMajor == 5 && kernelMinor >= 17)
}

func FixupSpec(spec *ebpf.CollectionSpec, overrideKernelVersion bool) {
	if !SupportsEBPFLoops(ptlog(), overrideKernelVersion) {
		// Hack: instead of redefining bpf2go generated struct for mutually exclusive conditional programs,
		// use one predefined field name to store either of them.
		spec.Programs["obi_protocol_http"] = spec.Programs["obi_protocol_http_legacy"]
		spec.Programs["obi_protocol_http"].Name = "obi_protocol_http"
		spec.Programs["obi_continue_protocol_http"] = spec.Programs["obi_continue_protocol_http_legacy"]
		spec.Programs["obi_continue_protocol_http"].Name = "obi_continue_protocol_http"
	}

	dummy := &ebpf.ProgramSpec{
		Name: "obi_dummy",
		Type: ebpf.Kprobe,
		Instructions: asm.Instructions{
			asm.Mov.Imm(asm.R0, 0),
			asm.Return(),
		},
		License: "MIT",
	}

	// Hack: insert dummy unused programs in order to be able to use bpf2go generated struct to load
	// the collection.
	spec.Programs["obi_protocol_http_legacy"] = dummy.Copy()
	spec.Programs["obi_continue_protocol_http_legacy"] = dummy.Copy()
}

// Injectable for tests
var lockdownPath = "/sys/kernel/security/lockdown"

func KernelLockdownMode() KernelLockdown {
	plog := ptlog()
	plog.Debug("checking kernel lockdown mode, [none] allows us to propagate trace context")
	// If we can't find the file, assume no lockdown
	if _, err := os.Stat(lockdownPath); err == nil {
		f, err := os.Open(lockdownPath)
		if err != nil {
			plog.Warn("failed to open /sys/kernel/security/lockdown, assuming lockdown [integrity]", "error", err)
			return KernelLockdownIntegrity
		}

		defer f.Close()
		scanner := bufio.NewScanner(f)
		if scanner.Scan() {
			lockdown := scanner.Text()
			switch {
			case strings.Contains(lockdown, "[none]"):
				return KernelLockdownNone
			case strings.Contains(lockdown, "[integrity]"):
				return KernelLockdownIntegrity
			case strings.Contains(lockdown, "[confidentiality]"):
				return KernelLockdownConfidentiality
			default:
				return KernelLockdownOther
			}
		}

		plog.Warn("file /sys/kernel/security/lockdown is empty, assuming lockdown [integrity]")
		return KernelLockdownIntegrity
	}

	plog.Debug("can't find /sys/kernel/security/lockdown, assuming no lockdown")
	return KernelLockdownNone
}

func cstr(chars []uint8) string {
	addrLen := bytes.IndexByte(chars, 0)
	if addrLen < 0 {
		addrLen = len(chars)
	}

	return string(chars[:addrLen])
}

func (connInfo *BPFConnInfo) reqHostInfo() (source, target string) {
	src := make(net.IP, net.IPv6len)
	dst := make(net.IP, net.IPv6len)
	copy(src, connInfo.S_addr[:])
	copy(dst, connInfo.D_addr[:])

	srcStr := src.String()
	dstStr := dst.String()

	if src.IsUnspecified() {
		srcStr = ""
	}

	if dst.IsUnspecified() {
		dstStr = ""
	}

	return srcStr, dstStr
}

func isClientEvent(et uint8) bool {
	switch request.EventType(et) {
	case request.EventTypeGRPCClient, request.EventTypeHTTPClient, request.EventTypeRedisClient,
		request.EventTypeKafkaClient, request.EventTypeNATSClient, request.EventTypeAMQPClient, request.EventTypeSQLClient, request.EventTypeMongoClient,
		request.EventTypeFailedConnect:
		return true
	}

	return false
}

func directionByPacketType(pt uint8, isClient bool) uint8 {
	if isClient {
		if pt == packetTypeRequest {
			return directionSend
		}
		return directionRecv
	}
	if pt == packetTypeRequest {
		return directionRecv
	}
	return directionSend
}