diff --git a/backend/app/agent_assets/nexafabric-agent.py b/backend/app/agent_assets/nexafabric-agent.py index e975b79..73dd8f2 100644 --- a/backend/app/agent_assets/nexafabric-agent.py +++ b/backend/app/agent_assets/nexafabric-agent.py @@ -22,6 +22,7 @@ VERSION = "0.3.0" VM_INTERFACE_RE = re.compile(r"(?:tap|fwbr|fwln|fwpr)(\d+)") VM_INTERFACE_DETAIL_RE = re.compile(r"(?:tap|fwbr|fwln|fwpr)(\d+)i(\d+)") LOG_FIELD_RE = re.compile(r"\b([A-Z]+)=([^\s]+)") +FIREWALL_LOG_TS_RE = re.compile(r"\b(\d{2}/[A-Za-z]{3}/\d{4}:\d{2}:\d{2}:\d{2} [+-]\d{4})\b") ETH_P_IP = 0x0800 ETH_P_ALL = 0x0003 ETH_P_8021Q = 0x8100 @@ -255,13 +256,18 @@ def collect_packet_flows(interfaces: list[dict[str, Any]], duration: int, limit: if current: current["packets"] += 1 current["bytes"] += int(flow["bytes"]) + current["last_seen_at"] = datetime.now(timezone.utc).isoformat() continue + now = datetime.now(timezone.utc).isoformat() flow.update( { "vmid": vmid, "nic": nic, "interface": name, "collector": "linux-af-packet", + "first_seen_at": now, + "last_seen_at": now, + "observed_at": now, } ) flows[key] = flow @@ -332,12 +338,27 @@ def collect_flows(limit: int = 500) -> list[dict[str, Any]]: if key in seen: continue seen.add(key) + now = datetime.now(timezone.utc).isoformat() + flow["first_seen_at"] = now + flow["last_seen_at"] = now + flow["observed_at"] = now + flow["collector"] = flow.get("collector") or "conntrack" flows.append(flow) if len(flows) >= limit: break return flows +def firewall_log_seen_at(line: str) -> str: + match = FIREWALL_LOG_TS_RE.search(line) + if match: + try: + return datetime.strptime(match.group(1), "%d/%b/%Y:%H:%M:%S %z").astimezone(timezone.utc).isoformat() + except ValueError: + pass + return datetime.now(timezone.utc).isoformat() + + def parse_firewall_log_line(line: str) -> dict[str, Any] | None: fields = {key: value for key, value in LOG_FIELD_RE.findall(line)} source_ip = fields.get("SRC") @@ -358,6 +379,7 @@ def parse_firewall_log_line(line: str) -> dict[str, Any] | None: source_port = fields.get("SPT") destination_port = fields.get("DPT") length = fields.get("LEN") + seen_at = firewall_log_seen_at(line) return { "source_ip": source_ip, "destination_ip": destination_ip, @@ -370,6 +392,9 @@ def parse_firewall_log_line(line: str) -> dict[str, Any] | None: "decision": decision, "collector": "firewall-log", "log_excerpt": line[-180:], + "first_seen_at": seen_at, + "last_seen_at": seen_at, + "observed_at": seen_at, } @@ -423,6 +448,8 @@ def collect_firewall_log_flows(since_minutes: int = 5, limit: int = 500) -> tupl if current: current["packets"] += 1 current["bytes"] += int(flow.get("bytes") or 0) + current["last_seen_at"] = flow.get("last_seen_at") or datetime.now(timezone.utc).isoformat() + current["observed_at"] = current["last_seen_at"] continue flows[key] = flow if len(flows) >= limit: @@ -448,6 +475,11 @@ def merge_flow_sources(*sources: list[dict[str, Any]], limit: int = 500) -> list if current: current["packets"] = int(current.get("packets") or 0) + int(flow.get("packets") or 0) current["bytes"] = int(current.get("bytes") or 0) + int(flow.get("bytes") or 0) + current_seen = str(current.get("last_seen_at") or current.get("observed_at") or "") + flow_seen = str(flow.get("last_seen_at") or flow.get("observed_at") or "") + if flow_seen > current_seen: + current["last_seen_at"] = flow_seen + current["observed_at"] = flow_seen continue flows[key] = dict(flow) return sorted( diff --git a/backend/app/agent_assets/nexafabric-ebpf.go b/backend/app/agent_assets/nexafabric-ebpf.go index 1c244eb..f00362d 100644 --- a/backend/app/agent_assets/nexafabric-ebpf.go +++ b/backend/app/agent_assets/nexafabric-ebpf.go @@ -55,13 +55,16 @@ type flowValue struct { Direction string `json:"direction,omitempty"` State string `json:"state"` Collector string `json:"collector"` + FirstSeenAt string `json:"first_seen_at"` + LastSeenAt string `json:"last_seen_at"` + ObservedAt string `json:"observed_at"` } type diagnostics struct { - AttachMode string `json:"attach_mode"` + AttachMode string `json:"attach_mode"` InterfacesRequested []string `json:"interfaces_requested"` InterfacesAttached []string `json:"interfaces_attached"` - Errors []string `json:"errors"` + Errors []string `json:"errors"` } type payload struct { @@ -189,9 +192,11 @@ func openSocket(interfaceName string) (int, error) { func collect(interfaceNames []string, duration time.Duration, limit int) payload { result := payload{ + Flows: []flowValue{}, Diagnostics: diagnostics{ AttachMode: "af_packet_raw_socket", InterfacesRequested: interfaceNames, + InterfacesAttached: []string{}, Errors: []string{}, }, } @@ -258,6 +263,7 @@ func collect(interfaceNames []string, duration time.Duration, limit int) payload if !ok { continue } + now := time.Now().UTC().Format(time.RFC3339Nano) key.Interface = socket.name key.VMID, key.NIC = interfaceMeta(socket.name) key.Direction = "ingress" @@ -278,11 +284,16 @@ func collect(interfaceNames []string, duration time.Duration, limit int) payload Direction: key.Direction, State: "observed", Collector: "ebpf-helper", + FirstSeenAt: now, + LastSeenAt: now, + ObservedAt: now, } flows[key] = current } current.Packets++ current.Bytes += uint64(bytes) + current.LastSeenAt = now + current.ObservedAt = now if len(flows) >= limit { break } diff --git a/backend/app/api/v1/router.py b/backend/app/api/v1/router.py index 0aff082..047f9a1 100644 --- a/backend/app/api/v1/router.py +++ b/backend/app/api/v1/router.py @@ -656,6 +656,20 @@ def traffic_flow_key(node_id: str, raw_flow: dict[str, object]) -> tuple[object, ) +def flow_observed_at(raw_flow: dict[str, object], payload_collected_at: datetime | None) -> datetime: + for key in ("last_seen_at", "observed_at", "first_seen_at"): + value = raw_flow.get(key) + if not value: + continue + try: + parsed = datetime.fromisoformat(str(value).replace("Z", "+00:00")) + except ValueError: + continue + return parsed.replace(tzinfo=None) if parsed.tzinfo else parsed + fallback = payload_collected_at or datetime.utcnow() + return fallback.replace(tzinfo=None) if fallback.tzinfo else fallback + + def compact_agent_payload(payload: AgentHeartbeat) -> dict: value = payload.model_dump(mode="json") flows = value.get("flows") @@ -1581,8 +1595,7 @@ def agent_heartbeat(payload: AgentHeartbeat, authorization: str | None = Header( destination_ip = str(raw_flow.get("destination_ip") or "") if not source_ip or not destination_ip: continue - observed_at = payload.collected_at or datetime.utcnow() - observed_at = observed_at.replace(tzinfo=None) if observed_at.tzinfo else observed_at + observed_at = flow_observed_at(raw_flow, payload.collected_at) key = traffic_flow_key(node.id, raw_flow) existing = existing_flows.get(key) if existing: diff --git a/frontend/src/pages/Workloads.tsx b/frontend/src/pages/Workloads.tsx index 2e5a9c8..f1de747 100644 --- a/frontend/src/pages/Workloads.tsx +++ b/frontend/src/pages/Workloads.tsx @@ -94,6 +94,12 @@ function summarizeTraffic(traffic: Array>) { } if (!existing.observedAt && flow.observed_at) { existing.observedAt = String(flow.observed_at); + } else if (flow.observed_at) { + const existingTime = Date.parse(existing.observedAt || ""); + const flowTime = Date.parse(String(flow.observed_at)); + if (Number.isFinite(flowTime) && (!Number.isFinite(existingTime) || flowTime > existingTime)) { + existing.observedAt = String(flow.observed_at); + } } continue; }