diff --git a/backend/app/agent_assets/EBPF_HELPER_CONTRACT.md b/backend/app/agent_assets/EBPF_HELPER_CONTRACT.md index 69b077a..8c3ebb2 100644 --- a/backend/app/agent_assets/EBPF_HELPER_CONTRACT.md +++ b/backend/app/agent_assets/EBPF_HELPER_CONTRACT.md @@ -1,6 +1,9 @@ # NexaFabric eBPF helper contract Agent 0.3.0 can call an optional helper binary at `/opt/nexafabric-agent/nexafabric-ebpf`. +The repository includes a dependency-free Go helper source at `nexafabric-ebpf.go`. + +The first implementation uses Linux raw packet sockets on the selected VM/LXC interfaces and prints the same JSON contract that a tc/eBPF implementation should print. This keeps the helper installable on Proxmox immediately while preserving the agent integration point for a later kernel eBPF loader. The helper is invoked as: @@ -28,7 +31,7 @@ It must print JSON to stdout: } ], "diagnostics": { - "attach_mode": "tc", + "attach_mode": "af_packet_raw_socket", "interfaces_attached": ["tap100i0"] } } diff --git a/backend/app/agent_assets/nexafabric-ebpf.go b/backend/app/agent_assets/nexafabric-ebpf.go new file mode 100644 index 0000000..f28e519 --- /dev/null +++ b/backend/app/agent_assets/nexafabric-ebpf.go @@ -0,0 +1,334 @@ +package main + +import ( + "encoding/binary" + "encoding/json" + "flag" + "fmt" + "net" + "os" + "regexp" + "sort" + "strings" + "syscall" + "time" +) + +const ( + ethPAll = 0x0003 + ethPIP = 0x0800 + ethP8021Q = 0x8100 + ethP8021AD = 0x88A8 + packetOut = 4 + protoICMP = 1 + protoTCP = 6 + protoUDP = 17 + maxFrameSize = 65535 +) + +var vmInterfaceRE = regexp.MustCompile(`(?:tap|fwbr|fwln|fwpr)(\d+)`) +var vmInterfaceDetailRE = regexp.MustCompile(`(?:tap|fwbr|fwln|fwpr)(\d+)i(\d+)`) + +type flowKey struct { + VMID string + NIC string + Interface string + Direction string + SourceIP string + DestinationIP string + Protocol string + SourcePort int + DestinationPort int +} + +type flowValue struct { + SourceIP string `json:"source_ip"` + DestinationIP string `json:"destination_ip"` + Protocol string `json:"protocol"` + SourcePort *int `json:"source_port,omitempty"` + DestinationPort *int `json:"destination_port,omitempty"` + Packets uint64 `json:"packets"` + Bytes uint64 `json:"bytes"` + VMID string `json:"vmid,omitempty"` + NIC string `json:"nic,omitempty"` + Interface string `json:"interface,omitempty"` + Direction string `json:"direction,omitempty"` + State string `json:"state"` + Collector string `json:"collector"` +} + +type diagnostics struct { + AttachMode string `json:"attach_mode"` + InterfacesRequested []string `json:"interfaces_requested"` + InterfacesAttached []string `json:"interfaces_attached"` + Errors []string `json:"errors"` +} + +type payload struct { + Flows []flowValue `json:"flows"` + Diagnostics diagnostics `json:"diagnostics"` +} + +func htons(value uint16) int { + return int((value<<8)&0xff00 | value>>8) +} + +func intPtr(value int) *int { + if value == 0 { + return nil + } + return &value +} + +func protocolName(value byte) string { + switch value { + case protoICMP: + return "icmp" + case protoTCP: + return "tcp" + case protoUDP: + return "udp" + default: + return "" + } +} + +func parsePacket(packet []byte) (flowKey, int, bool) { + var key flowKey + if len(packet) < 34 { + return key, 0, false + } + offset := 12 + ethType := binary.BigEndian.Uint16(packet[offset : offset+2]) + offset = 14 + for ethType == ethP8021Q || ethType == ethP8021AD { + if len(packet) < offset+4 { + return key, 0, false + } + ethType = binary.BigEndian.Uint16(packet[offset+2 : offset+4]) + offset += 4 + } + if ethType != ethPIP || len(packet) < offset+20 { + return key, 0, false + } + versionIHL := packet[offset] + version := versionIHL >> 4 + ihl := int(versionIHL&0x0f) * 4 + if version != 4 || ihl < 20 || len(packet) < offset+ihl { + return key, 0, false + } + totalLength := int(binary.BigEndian.Uint16(packet[offset+2 : offset+4])) + protocol := protocolName(packet[offset+9]) + if protocol == "" { + return key, 0, false + } + key.SourceIP = net.IP(packet[offset+12 : offset+16]).String() + key.DestinationIP = net.IP(packet[offset+16 : offset+20]).String() + key.Protocol = protocol + transportOffset := offset + ihl + if protocol == "tcp" || protocol == "udp" { + if len(packet) < transportOffset+4 { + return key, 0, false + } + key.SourcePort = int(binary.BigEndian.Uint16(packet[transportOffset : transportOffset+2])) + key.DestinationPort = int(binary.BigEndian.Uint16(packet[transportOffset+2 : transportOffset+4])) + } else if protocol == "icmp" && len(packet) >= transportOffset+2 { + key.SourcePort = int(packet[transportOffset]) + key.DestinationPort = int(packet[transportOffset+1]) + } + if totalLength <= 0 { + totalLength = len(packet) - offset + } + return key, totalLength, true +} + +func interfaceMeta(name string) (string, string) { + vmid := "" + nic := "0" + if match := vmInterfaceRE.FindStringSubmatch(name); len(match) > 1 { + vmid = match[1] + } + if match := vmInterfaceDetailRE.FindStringSubmatch(name); len(match) > 2 { + nic = match[2] + } + return vmid, nic +} + +func setFd(fd int, set *syscall.FdSet) { + set.Bits[fd/64] |= 1 << uint(fd%64) +} + +func isSet(fd int, set *syscall.FdSet) bool { + return set.Bits[fd/64]&(1< maxFd { + maxFd = socket.fd + } + } + flows := map[flowKey]*flowValue{} + deadline := time.Now().Add(duration) + buffer := make([]byte, maxFrameSize) + for time.Now().Before(deadline) { + var readfds syscall.FdSet + for _, socket := range sockets { + setFd(socket.fd, &readfds) + } + timeout := syscall.NsecToTimeval(int64(250 * time.Millisecond)) + _, err := syscall.Select(maxFd+1, &readfds, nil, nil, &timeout) + if err != nil && err != syscall.EINTR { + result.Diagnostics.Errors = append(result.Diagnostics.Errors, err.Error()) + break + } + for fd, socket := range fdToSocket { + if !isSet(fd, &readfds) { + continue + } + n, from, err := syscall.Recvfrom(fd, buffer, 0) + if err != nil { + if err != syscall.EAGAIN && err != syscall.EWOULDBLOCK { + result.Diagnostics.Errors = append(result.Diagnostics.Errors, fmt.Sprintf("%s: %v", socket.name, err)) + } + continue + } + key, bytes, ok := parsePacket(buffer[:n]) + if !ok { + continue + } + key.Interface = socket.name + key.VMID, key.NIC = interfaceMeta(socket.name) + key.Direction = "ingress" + if link, ok := from.(*syscall.SockaddrLinklayer); ok && link.Pkttype == packetOut { + key.Direction = "egress" + } + current := flows[key] + if current == nil { + current = &flowValue{ + SourceIP: key.SourceIP, + DestinationIP: key.DestinationIP, + Protocol: key.Protocol, + SourcePort: intPtr(key.SourcePort), + DestinationPort: intPtr(key.DestinationPort), + VMID: key.VMID, + NIC: key.NIC, + Interface: key.Interface, + Direction: key.Direction, + State: "observed", + Collector: "ebpf-helper", + } + flows[key] = current + } + current.Packets++ + current.Bytes += uint64(bytes) + if len(flows) >= limit { + break + } + } + if len(flows) >= limit { + break + } + } + for _, flow := range flows { + result.Flows = append(result.Flows, *flow) + } + sort.Slice(result.Flows, func(i, j int) bool { + if result.Flows[i].Bytes == result.Flows[j].Bytes { + return result.Flows[i].Packets > result.Flows[j].Packets + } + return result.Flows[i].Bytes > result.Flows[j].Bytes + }) + if len(result.Flows) > limit { + result.Flows = result.Flows[:limit] + } + return result +} + +func main() { + jsonOutput := flag.Bool("json", false, "print JSON output") + limit := flag.Int("limit", 1500, "maximum unique flows") + durationSeconds := flag.Int("duration", 10, "collection duration in seconds") + interfaces := flag.String("interfaces", "", "comma-separated interface names") + flag.Parse() + if !*jsonOutput { + fmt.Fprintln(os.Stderr, "only --json output is supported") + os.Exit(2) + } + if *limit <= 0 { + *limit = 1500 + } + if *durationSeconds <= 0 { + *durationSeconds = 10 + } + names := []string{} + for _, name := range strings.Split(*interfaces, ",") { + if strings.TrimSpace(name) != "" { + names = append(names, strings.TrimSpace(name)) + } + } + result := collect(names, time.Duration(*durationSeconds)*time.Second, *limit) + data, err := json.Marshal(result) + if err != nil { + fmt.Fprintf(os.Stderr, "json marshal failed: %v\n", err) + os.Exit(1) + } + fmt.Println(string(data)) +} diff --git a/backend/app/api/v1/router.py b/backend/app/api/v1/router.py index 018d6bf..0aff082 100644 --- a/backend/app/api/v1/router.py +++ b/backend/app/api/v1/router.py @@ -1410,9 +1410,19 @@ if ! command -v conntrack >/dev/null 2>&1 && command -v apt-get >/dev/null 2>&1; apt-get install -y conntrack fi +if ! command -v go >/dev/null 2>&1 && command -v apt-get >/dev/null 2>&1; then + apt-get update + apt-get install -y golang-go || true +fi + mkdir -p "$INSTALL_DIR" "$CONFIG_DIR" curl -fsSL "$NEXAFABRIC_URL/api/v1/agents/download/nexafabric-agent.py" -o "$INSTALL_DIR/nexafabric-agent.py" chmod 0755 "$INSTALL_DIR/nexafabric-agent.py" +curl -fsSL "$NEXAFABRIC_URL/api/v1/agents/download/nexafabric-ebpf.go" -o "$INSTALL_DIR/nexafabric-ebpf.go" || true +if command -v go >/dev/null 2>&1 && [ -f "$INSTALL_DIR/nexafabric-ebpf.go" ]; then + (cd "$INSTALL_DIR" && go build -o "$INSTALL_DIR/nexafabric-ebpf" "$INSTALL_DIR/nexafabric-ebpf.go") || true + [ -f "$INSTALL_DIR/nexafabric-ebpf" ] && chmod 0755 "$INSTALL_DIR/nexafabric-ebpf" +fi cat > "$CONFIG_DIR/config.json" <<'JSON' {{ @@ -1522,6 +1532,12 @@ def download_node_agent() -> FileResponse: return FileResponse(path, media_type="text/x-python", filename="nexafabric-agent.py") +@api_router.get("/agents/download/nexafabric-ebpf.go") +def download_node_agent_ebpf_helper() -> FileResponse: + path = Path(__file__).resolve().parents[2] / "agent_assets" / "nexafabric-ebpf.go" + return FileResponse(path, media_type="text/x-go", filename="nexafabric-ebpf.go") + + @api_router.post("/agents/heartbeat") def agent_heartbeat(payload: AgentHeartbeat, authorization: str | None = Header(default=None), db: Session = Depends(get_db)) -> dict: if not authorization or not authorization.lower().startswith("bearer "):