ws support in ebpf

[kural-akto]

WebSocket Support in eBPF Traffic Capture

This PR adds WebSocket protocol detection and message capture to the eBPF-based traffic interception layer.

What Changed

The eBPF agent previously only captured HTTP/1.x and egress traffic. This PR extends it to detect WebSocket upgrade handshakes and route subsequent binary frames through a dedicated WebSocket pipeline.

---

Flow Diagram

  Kernel eBPF Layer
  ─────────────────
  TCP data events (send/recv buffers)
         │
         ▼
  ProcessTrackerData()   [factory.go]
         │
         ├─── Already a known WebSocket conn? ──YES──► processWebSocketConnection()
         │                                                    │
         │                                                    ├── Parse binary WS frames
         │                                                    ├── Accumulate in WSConnectionManager
         │                                                    └── Flush to Kafka on close frame / completion
         │
         └─── NO: isWebSocketUpgradeData(sent, recv)?
                    │
              ┌─────┴─────┐
             YES           NO
              │             │
              ▼             ▼
   markAsWebSocketConnection()    Standard HTTP path
   RegisterConnectionNumeric()    tryReadFromBD() → Kafka
              │
              ▼
   Strip WS frames from buffers
   Send HTTP handshake only to tryReadFromBD()
   (101 Switching Protocols captured as normal HTTP pair)


  WebSocket Frame Path (subsequent data after upgrade)
  ──────────────────────────────────────────────────────

  raw bytes
     │
     ▼
  parseWebSocketFrames()      [factory.go]
  - Reads FIN, opcode, mask bit, payload length
  - Handles 16-bit / 64-bit extended length
  - Unmasks payload if masked
     │
     ▼
  []WebSocketMessage  { Payload, Direction, Opcode, FIN, Masked, Timestamp }
     │
     ├── hasCloseFrame? ──YES──► flush batch + unmarkWebSocketConnection()
     │
     └── AccumulateMessagesNumeric()  [websocket_manager.go]
             │
             ▼
         WSConnectionManager (in-memory, keyed by ConnID.Id:ConnID.Fd)
             │
             ▼ (on flush)
         ProduceStr() → Kafka topic

---

Key Design Decisions

Decision	Rationale
Skip sequence checks for WS connections	Binary WebSocket frames don't follow sequential HTTP packet numbering, so convertToSingleByteArr now accepts isWebsocket bool to bypass sequence validation
Two-phase detection	First pass detects HTTP 101 upgrade (headers check), marks the connID, then all future data on that connID is routed as WebSocket frames
Numeric connID key (Id:Fd)	Stable kernel-level identifier; avoids IP string parsing issues for the eBPF path
HTTP handshake captured separately	The initial GET /ws + 101 Switching Protocols pair is stripped of WS frames and sent through the normal HTTP pipeline so the upgrade request appears in API inventory
Stale connection cleanup	Background goroutine every 5 minutes evicts connections idle for > 30 minutes

---

Files Changed

• ebpf/connections/factory.go — WebSocket detection, frame parsing, connection lifecycle management
• trafficUtil/kafkaUtil/websocket_manager.go (new) — Thread-safe WebSocketConnectionManager with accumulate/flush/cleanup API

[gauravakto]

Code Review

Bugs / Correctness Issues

1. Double lock in ProcessTrackerData — race-prone and redundant

isKnownWebSocket := isWebSocketConnection(connID)  // acquires RLock
// ...
if isWebSocketConnection(connID) {                  // acquires RLock again

The state could change between the two calls. isKnownWebSocket is computed but the branch uses a second fresh call. Use isKnownWebSocket consistently.

---

2. wsConnections map never cleaned up on TCP drop — memory leak

unmarkWebSocketConnection is only called when a WebSocket close frame (opcode 8) is detected. If the TCP connection drops without a clean WS close frame (network failure, client crash), the entry stays in wsConnections forever. The two empty no-op blocks added in DeleteWorker make this explicit:

if _, ok := factory.connections[connectionID]; ok {
    // Don't unmark WebSocket connections...
}

WSConnectionManager has TTL-based cleanup but wsConnections (the map[string]bool in factory.go) has none. Long-running services will leak entries.

---

3. ParseWebSocketFrames mutates the caller's input slice

framePayload := payload[i : i+payloadLen]  // slice into original buffer — no copy
if masked {
    for j := 0; j < len(framePayload); j++ {
        framePayload[j] ^= maskingKey[j%4]  // mutates original payload!
    }
}

Fix: copy before unmasking:

framePayload := make([]byte, payloadLen)
copy(framePayload, payload[i:i+payloadLen])

---

4. 64-bit payload length silently truncated to int

payloadLen = int(payload[i])<<56 | ...

On 32-bit systems this overflows. RFC 6455 says the top bit must be 0, but the code should still guard:

if payloadLen < 0 || payloadLen > someMaxSafeSize {
    break
}

---

5. :9092 Kafka exclusion in isWebSocketUpgradeData is fragile

Checking bytes.Contains(sentBuffer, []byte(":9092")) on raw TCP bytes will incorrectly exclude any WebSocket connection whose payload happens to contain the string :9092. The port check should happen at the connection level (connID.Port), not by scanning the data buffer.

---

6. Hardcoded akto_account_id: 1000000 in two places

Appears in both sendWebSocketBatches and the handshake payload in parser.go. Should come from context/config.

---

7. sendWebSocketBatches parses the key string it constructed

parts := strings.Split(key, ":")  // re-parsing "%d:%d"

Tight coupling between key format and the consumer. If the format changes, the batch sender silently breaks. Better to expose a typed iterator from WSConnectionManager.

---

Design Issues

8. Two parallel WebSocket tracking systems that can desync

• wsConnections map[string]bool in factory.go — tracks whether a connID is WebSocket
• WSConnectionManager in websocket_manager.go — stores messages and metadata

Both keyed by "Id:Fd" but managed separately. Only the manager has TTL cleanup. They can get out of sync if one is cleaned up but not the other. A single source of truth would eliminate the class of bugs.

---

9. Two registration paths store IPs differently

• RegisterConnection (string path, called from PCAP/mirroring in ParseAndProduce) stores real IP strings
• RegisterConnectionNumeric (eBPF path) stores raw integer values as decimal strings: fmt.Sprintf("%d", srcIPNumeric)

The downstream Kafka payload will look completely different between the two paths for the same field names (ip, destIp). The numeric path should convert to dotted-quad format.

---

10. Messages accumulate unboundedly between 1-minute flushes

AccumulateMessagesNumeric appends without limit. A high-throughput WebSocket connection (e.g. streaming financial data) could accumulate tens of thousands of frames in memory between flushes. Need a cap or size-based flush trigger.

---

Minor Issues

• Empty no-op blocks in DeleteWorker — two if _, ok := ...; ok { // comment } blocks that do nothing should be removed
• slog.Info in hot path — markAsWebSocketConnection / unmarkWebSocketConnection fire on every upgrade/close; should be slog.Debug
• Duplicate detection logic — isWebSocketUpgradeData (raw bytes, factory.go) and detectWebSocketUpgrade (http.Response, parser.go) both detect WS upgrades with different implementations
• Binary payloads stored as string — Payload: string(framePayload) for opcode 2 (binary) frames produces invalid UTF-8. Consider base64-encoding binary frames
• Blank line inside extractHTTPHeadersOnly after if headerEnd == -1 { is noise

---

Summary

Severity	Issue
Bug	In-place buffer mutation in frame parser (data corruption)
Bug	wsConnections map leak on unclean TCP close
Bug	Double isWebSocketConnection call with inconsistent state window
Bug	Hardcoded account ID
Design	Dual tracking systems can desync
Design	Unbounded message accumulation
Design	Numeric IPs stored as raw integers, not dotted-quad

The core detection logic is sound. The in-place buffer mutation is the highest priority fix — it will cause subtle data corruption on masked frames. The wsConnections leak and dual tracking system are the main architectural concerns.

[gauravakto]

Strict Code Review — WebSocket Support in eBPF

---

🔴 Critical Bugs

1. Silent close-frame detection failure — connections never cleaned up in practice

ParseWebSocketFrames skips frames where opcode >= 0x8 && payloadLen == 0:

if opcode >= 0x8 && payloadLen == 0 {
    continue  // skipped — never appended to messages
}
messages = append(messages, msg)

RFC 6455 §5.5 explicitly allows a close frame with no status code (empty payload). Most WebSocket clients (browsers, Go, Python) send exactly this form on a clean shutdown. Because the frame is skipped, hasCloseFrame(frames) never returns true, unmarkWebSocketConnection is never called, and every WebSocket connection leaks permanently regardless of whether it closes cleanly. The TTL-based cleanup in WSConnectionManager will eventually evict the manager entry, but wsConnections (the map[string]bool in factory.go) has no TTL and will grow unboundedly.

Fix: append control frames to the returned slice before deciding whether to skip them for Kafka production, or detect them separately before the skip.

---

2. ParseWebSocketFrames mutates the caller's buffer — data corruption

framePayload := payload[i : i+payloadLen]  // subslice — shares backing array
if masked {
    for j := range framePayload {
        framePayload[j] ^= maskingKey[j%4]  // writes into caller's payload
    }
}

The caller in processWebSocketConnection passes sentBuffer or receiveBuffer directly. These buffers are the assembled eBPF tracker buffers. Mutating them corrupts the original data for any subsequent use. All browser-to-server frames are masked (RFC 6455 §5.3 requires it), so this path is always hit for ingress frames.

// Fix:
framePayload := make([]byte, payloadLen)
copy(framePayload, payload[i:i+payloadLen])

---

3. Unbounded allocation from attacker-controlled payload length

payloadLen = int(payload[i])<<56 | int(payload[i+1])<<48 | ...
// ...
framePayload := payload[i : i+payloadLen]  // bounds-checked by `i+payloadLen > len(payload)` only

The bounds check prevents an out-of-bounds slice panic, but there is no cap on payloadLen before the slice is taken. A valid-looking frame with a length of len(payload) - i will silently process a multi-megabyte chunk as a single message. Combined with unbounded accumulation in the manager, this enables memory exhaustion from a single connection. Add a per-frame maximum (e.g. 1 MB) and drop or truncate frames exceeding it.

Additionally, on 32-bit architectures the 64-bit shift into int overflows silently. Use uint64 for the intermediate calculation and bounds-check before narrowing to int.

---

4. TOCTOU race on WebSocket connection state

isKnownWebSocket := isWebSocketConnection(connID)   // RLock, returns bool, releases lock
receiveBuffer := convertToSingleByteArr(tracker.recvBuf, isKnownWebSocket)
sentBuffer    := convertToSingleByteArr(tracker.sentBuf, isKnownWebSocket)

if isWebSocketConnection(connID) {   // acquires RLock again — state may have changed
    processWebSocketConnection(...)
    return
}

Between the two calls, another goroutine may have called markAsWebSocketConnection. The first call determines the sequence-check mode; the second call determines the routing. If state changes between them, a connection is routed as HTTP with WS sequence semantics, or vice versa. Use isKnownWebSocket for both.

---

5. Non-atomic cleanup across two independent locks

// processWebSocketConnection, on close frame:
unmarkWebSocketConnection(connID)                          // Lock A
kafkaUtil.WSConnectionManager.RemoveConnectionByConnID(...)  // Lock B

These are two separate mutexes. A concurrent ProcessTrackerData call between the two unlocks sees a state where wsConnections no longer has the key (so it's treated as HTTP) but WSConnectionManager still has a live entry with accumulated messages. Any data arriving in that window is silently dropped.

---

🟠 Correctness Issues

6. WebSocket message fragmentation not handled — RFC 6455 §5.4 violated

The parser treats each frame independently. Fragmented messages have:

• First frame: opcode = 1 (text) or 2 (binary), FIN = 0
• Continuation frames: opcode = 0x0, FIN = 0
• Final frame: opcode = 0x0, FIN = 1

The code ignores the FIN bit and opcode 0 continuation frames entirely. Fragmented messages are emitted as multiple disconnected payloads with the wrong opcode on continuation frames (0 instead of the original type). Any consumer trying to interpret the Kafka output will see malformed messages for any fragmented WebSocket exchange.

---

7. RSV bits not validated — extension negotiation silently broken

RFC 6455 §5.2: RSV1/RSV2/RSV3 must be 0 unless a negotiated extension defines their meaning. The most common extension is permessage-deflate (RFC 7692), which sets RSV1 on compressed frames. The parser does not check RSV bits:

byte0 := payload[i]
fin    := (byte0 & 0x80) != 0
// RSV1 = byte0 & 0x40, RSV2 = byte0 & 0x20, RSV3 = byte0 & 0x10 — all ignored
opcode := byte0 & 0x0F

If permessage-deflate is in use (common in browser connections), frames with RSV1=1 are parsed as-is, and the payload is treated as raw text/binary when it is actually DEFLATE-compressed. The stored messages are garbage. Connections using compression should either be decoded or flagged.

---

8. Numeric IPs stored as raw integers in Kafka payload

RegisterConnectionNumeric stores:

SourceIP: fmt.Sprintf("%d", srcIPNumeric),  // e.g. "3232235777" for 192.168.1.1

The Kafka payload emitted by sendWebSocketBatches puts this in the "ip" field. All other traffic sources emit dotted-quad strings. Consumers downstream will receive inconsistent formats for the same field name. Convert using net.IP:

ip := make(net.IP, 4)
binary.LittleEndian.PutUint32(ip, srcIPNumeric)
SourceIP: ip.String()

---

9. akto_vxlan_id and source always hardcoded in batch sender

"akto_vxlan_id": fmt.Sprint(0),
"source":         "MIRRORING",

WebSocketConnection does not store the vxlan ID or traffic source from the original connection context. For PCAP or other sources, this will always emit incorrect metadata. These fields must be stored at registration time from TrafficContext.

---

10. Kafka akto_account_id hardcoded as 1000000

"akto_account_id": fmt.Sprint(1000000),

Appears in both sendWebSocketBatches (kafka.go) and the handshake payload (parser.go). This should be read from the same config/env source used by the rest of the pipeline.

---

11. All frames in a buffer get time.Now() as timestamp

msg := WebSocketMessage{
    Timestamp: time.Now(),  // time of processing, not time of capture

All frames parsed from a single receiveBuffer or sentBuffer receive identical timestamps — the moment the Go code runs. eBPF provides ktime_get_ns in the event structs for accurate capture timestamps. At minimum, frames within a buffer should have their order preserved; giving them all the same time makes ordering within a batch ambiguous.

---

🟡 Design Issues

12. Two independent tracking systems with no shared invariant

wsConnections map[string]bool (factory.go) and WSConnectionManager (websocket_manager.go) are both keyed by "Id:Fd" but managed through separate mutexes with no transaction boundary. There is no guarantee they are consistent at any point. Every operation that needs both must touch two locks, and the correctness of the overall system depends on an implicit ordering that is not enforced anywhere. Consolidate: either wsConnections checks WSConnectionManager.Has(key), or remove wsConnections and add an IsWebSocket(key) bool method to the manager.

---

13. sendWebSocketBatches parses the key string it constructed

connections := WSConnectionManager.GetAllConnections()  // returns []string keys
for _, key := range connections {
    parts := strings.Split(key, ":")  // re-parses "%d:%d"
    connIDId, _ := strconv.ParseUint(parts[0], 10, 64)
    connIDFd, _ := strconv.ParseUint(parts[1], 10, 32)
    batch, _ := WSConnectionManager.GetAndClearBatchByConnID(connIDId, uint32(connIDFd))
}

The batch sender inverts the key construction done in RegisterConnectionNumeric. If the key format ever changes (e.g. to add a namespace), the batch sender silently misparses. Expose GetAndClearBatch(key string) directly so the key is treated as an opaque handle.

---

14. GetAndClearBatch embeds stale Messages in the returned Connection copy

batch := &WebSocketBatch{
    Connection: *conn,       // value copy — Connection.Messages points to old backing array
    Messages:   conn.Messages,
}
conn.Messages = []WebSocketMessage{}

batch.Connection.Messages and batch.Messages both reference the same pre-reset slice. After this function, batch.Connection.Messages contains the full message list while the intent is for the connection to be cleared. Any consumer reading batch.Connection.Messages instead of batch.Messages sees the full unsent batch, potentially causing double-processing. Explicitly nil out Connection.Messages in the returned struct.

---

15. Dead code blocks in DeleteWorker

if _, ok := factory.connections[connectionID]; ok {
    // Don't unmark WebSocket connections — they maintain their type...
}

This block (and the identical one below) reads a value and does nothing. It adds noise and confusion about whether cleanup was intentional or forgotten. Remove both blocks; add a comment at the delete() call if the intent needs to be explained.

---

16. Duplicate WebSocket upgrade detection with divergent implementations

• isWebSocketUpgradeData in factory.go: raw byte scan for header strings and 101 Switching Protocols
• detectWebSocketUpgrade in parser.go: uses http.Response struct

These are two independent implementations of the same check. The raw-byte version has case-sensitivity gaps (checks Upgrade: websocket and upgrade: websocket but not UPGRADE: WEBSOCKET). Share one canonical implementation.

---

🔵 Minor Issues

• slog.Info in markAsWebSocketConnection/unmarkWebSocketConnection fires on every connection lifecycle event; should be slog.Debug to avoid flooding logs at scale
• isWebSocketUpgradeData scans for :9092 in raw data bytes — a payload containing the string :9092 will prevent WebSocket detection on that connection; port filtering belongs at the connID level
• Binary frame payloads (opcode 2) stored as string — produces invalid UTF-8; base64-encode binary frames before serialisation
• RegisterConnection (string path) is called from ParseAndProduce but never feeds frames — the PCAP/mirroring path registers connections in the manager but processWebSocketConnection (which feeds frames) is only reachable from the eBPF path; the manager entries from RegisterConnection accumulate with zero messages and are only evicted by TTL
• No unit tests for ParseWebSocketFrames, isWebSocketUpgradeData, extractHTTPHeadersOnly, or extractHeadersFromHTTPBuffer — these are pure functions with well-defined inputs and outputs; test coverage is straightforward and necessary given the parsing complexity

---

Summary Table

#	Severity	Issue
1	🔴 Critical	Close frame with empty payload never detected → all connections leak
2	🔴 Critical	In-place buffer mutation → data corruption on masked frames
3	🔴 Critical	Unbounded allocation from attacker-controlled payload length
4	🔴 Critical	TOCTOU race on WebSocket state check
5	🔴 Critical	Non-atomic dual-lock cleanup → data loss window
6	🟠 Correctness	Fragmented messages (opcode 0) not reassembled — RFC 6455 §5.4
7	🟠 Correctness	RSV bits ignored — compressed frames decoded as raw data
8	🟠 Correctness	Numeric IPs emitted as decimal integers, not dotted-quad
9	🟠 Correctness	vxlan ID and source hardcoded, not from connection context
10	🟠 Correctness	akto_account_id hardcoded as 1000000 in two places
11	🟠 Correctness	All frames get time.Now() — capture timestamps lost
12	🟡 Design	Two independent tracking maps with no shared invariant
13	🟡 Design	Batch sender re-parses opaque key string
14	🟡 Design	GetAndClearBatch returns stale Connection.Messages in value copy
15	🟡 Design	Dead no-op blocks in DeleteWorker
16	🟡 Design	Duplicate upgrade detection with divergent implementations
—	🔵 Minor	Log levels, binary payload encoding, dead registration path, no tests