From 977a2f4f56384f0844bc3c2be553be16c2db868f Mon Sep 17 00:00:00 2001 From: Aleksei Fedorov Date: Fri, 3 Apr 2026 18:10:09 +0000 Subject: [PATCH] python refactoring stage 2 --- kernel_ai/api/rest.py | 10 + kernel_ai/services/__init__.py | 2 + kernel_ai/services/network.py | 446 +++++++++ kernel_ai/services/processes.py | 543 +++++++++++ kernel_ai/services/system_view.py | 321 +++++++ kernel_ai/webapp.py | 1489 +---------------------------- 6 files changed, 1364 insertions(+), 1447 deletions(-) create mode 100644 kernel_ai/services/__init__.py create mode 100644 kernel_ai/services/network.py create mode 100644 kernel_ai/services/processes.py create mode 100644 kernel_ai/services/system_view.py diff --git a/kernel_ai/api/rest.py b/kernel_ai/api/rest.py index 283e480..24ed2c2 100644 --- a/kernel_ai/api/rest.py +++ b/kernel_ai/api/rest.py @@ -130,3 +130,13 @@ def processes_realtime(): @bp.route("/frontend-logs", methods=["POST", "OPTIONS"]) def ingest_frontend_logs(): return _core().ingest_frontend_logs() + + +@bp.route("/proc-graph") +def proc_graph(): + return _core().get_proc_graph() + + +@bp.route("/process-files") +def process_files(): + return _core().get_process_files() diff --git a/kernel_ai/services/__init__.py b/kernel_ai/services/__init__.py new file mode 100644 index 0000000..2eea431 --- /dev/null +++ b/kernel_ai/services/__init__.py @@ -0,0 +1,2 @@ +"""Domain services package.""" + diff --git a/kernel_ai/services/network.py b/kernel_ai/services/network.py new file mode 100644 index 0000000..fe13705 --- /dev/null +++ b/kernel_ai/services/network.py @@ -0,0 +1,446 @@ +"""Network domain service extracted from ``webapp``.""" + +from __future__ import annotations + +import ipaddress +import os +import re +import shutil +import subprocess +import time +from datetime import datetime + +import psutil + +from kernel_ai.state import NETWORK_STACK_PREV, TRACEROUTE_CACHE, TRACEROUTE_CACHE_TTL_SECONDS + + +def resolve_binary(cmd_name): + found = shutil.which(cmd_name) + if found: + return found + for base in ("/usr/sbin", "/usr/bin", "/sbin", "/bin"): + candidate = os.path.join(base, cmd_name) + if os.path.isfile(candidate) and os.access(candidate, os.X_OK): + return candidate + return None + + +def get_active_connections(): + """Get active network connections.""" + try: + connections = [] + with open("/proc/net/tcp", "r", encoding="utf-8", errors="ignore") as f: + lines = f.readlines()[1:] + for line in lines: + parts = line.strip().split() + if len(parts) < 4: + continue + local_addr = parts[1] + remote_addr = parts[2] + state = parts[3] + + def hex_to_ip(hex_str): + hex_bytes = [hex_str[i : i + 2] for i in range(0, 8, 2)] + hex_bytes.reverse() + return ".".join([str(int(b, 16)) for b in hex_bytes]) + + local_ip = hex_to_ip(local_addr.split(":")[0]) + local_port = int(local_addr.split(":")[1], 16) + + if remote_addr != "00000000:0000": + remote_ip = hex_to_ip(remote_addr.split(":")[0]) + remote_port = int(remote_addr.split(":")[1], 16) + connections.append( + { + "local": f"{local_ip}:{local_port}", + "remote": f"{remote_ip}:{remote_port}", + "state": state, + "type": "TCP", + } + ) + return connections[:20] + except Exception: + return get_mock_active_connections() + + +def get_mock_active_connections(): + return [ + {"local": "127.0.0.1:22", "remote": "192.168.1.100:54321", "state": "01", "type": "TCP"}, + {"local": "0.0.0.0:80", "remote": "10.0.0.50:12345", "state": "01", "type": "TCP"}, + {"local": "127.0.0.1:3306", "remote": "172.16.0.10:65432", "state": "01", "type": "TCP"}, + {"local": "0.0.0.0:443", "remote": "203.0.113.0:54321", "state": "01", "type": "TCP"}, + {"local": "127.0.0.1:5001", "remote": "192.168.1.101:12345", "state": "01", "type": "TCP"}, + ] + + +def _tcp_state_name(code): + states = { + "01": "ESTABLISHED", + "02": "SYN_SENT", + "03": "SYN_RECV", + "04": "FIN_WAIT1", + "05": "FIN_WAIT2", + "06": "TIME_WAIT", + "07": "CLOSE", + "08": "CLOSE_WAIT", + "09": "LAST_ACK", + "0A": "LISTEN", + "0B": "CLOSING", + } + return states.get(str(code).upper(), str(code).upper()) + + +def _get_default_iface(): + try: + with open("/proc/net/route", "r", encoding="utf-8", errors="ignore") as f: + lines = f.readlines()[1:] + for line in lines: + parts = line.strip().split() + if len(parts) < 11: + continue + iface = parts[0] + destination = parts[1] + flags = int(parts[3], 16) + if destination == "00000000" and (flags & 0x2): + return iface + except (OSError, ValueError): + pass + try: + pernic = psutil.net_io_counters(pernic=True) + for iface in pernic.keys(): + if iface != "lo": + return iface + except Exception: + pass + return "lo" + + +def _parse_netstat_tcpext(): + try: + with open("/proc/net/netstat", "r", encoding="utf-8", errors="ignore") as f: + lines = [line.strip() for line in f if line.strip()] + for i in range(0, len(lines) - 1, 2): + header = lines[i].split() + values = lines[i + 1].split() + if not header or header[0] != "TcpExt:": + continue + if not values or values[0] != "TcpExt:": + continue + fields = header[1:] + nums = values[1:] + if len(fields) != len(nums): + continue + mapping = {} + for name, val in zip(fields, nums): + try: + mapping[name] = int(val) + except ValueError: + mapping[name] = 0 + return mapping + except OSError: + return {} + return {} + + +def _parse_snmp_section(section_name): + try: + with open("/proc/net/snmp", "r", encoding="utf-8", errors="ignore") as f: + lines = [line.strip() for line in f if line.strip()] + for i in range(0, len(lines) - 1, 2): + header = lines[i].split() + values = lines[i + 1].split() + expected_prefix = f"{section_name}:" + if not header or header[0] != expected_prefix: + continue + if not values or values[0] != expected_prefix: + continue + fields = header[1:] + nums = values[1:] + if len(fields) != len(nums): + continue + out = {} + for name, val in zip(fields, nums): + try: + out[name] = int(val) + except ValueError: + out[name] = 0 + return out + except OSError: + return {} + return {} + + +def _get_ss_tcp_metrics(): + ss_cmd = resolve_binary("ss") + if not ss_cmd: + return {} + try: + result = subprocess.run([ss_cmd, "-tin"], capture_output=True, text=True, timeout=2, check=False) + lines = (result.stdout or "").splitlines() + except (subprocess.TimeoutExpired, OSError): + return {} + + for idx, line in enumerate(lines): + if not line.strip().startswith("ESTAB"): + continue + metrics = {} + parts = line.split() + if len(parts) >= 4: + try: + metrics["tx_queue"] = int(parts[2]) + metrics["rx_queue"] = int(parts[1]) + except ValueError: + pass + + details = lines[idx + 1] if (idx + 1) < len(lines) else "" + rtt_match = re.search(r"rtt:(\d+(?:\.\d+)?)/", details) + cwnd_match = re.search(r"cwnd:(\d+)", details) + retrans_match = re.search(r"retrans:(\d+)(?:/\d+)?", details) + if rtt_match: + metrics["rtt_ms"] = float(rtt_match.group(1)) + if cwnd_match: + metrics["cwnd"] = int(cwnd_match.group(1)) + if retrans_match: + metrics["retrans_now"] = int(retrans_match.group(1)) + if metrics: + return metrics + return {} + + +def get_network_stack_realtime(): + now = time.time() + iface = _get_default_iface() + pernic = psutil.net_io_counters(pernic=True) + iface_stats = pernic.get(iface) + all_connections = get_active_connections() + interesting = [c for c in all_connections if not c["remote"].startswith("127.0.0.1") and not c["remote"].startswith("0.0.0.0")] + flow = interesting[0] if interesting else (all_connections[0] if all_connections else None) + if flow: + flow = { + "local": flow.get("local"), + "remote": flow.get("remote"), + "type": str(flow.get("type", "TCP")).upper(), + "state_code": flow.get("state", "00"), + "state_name": _tcp_state_name(flow.get("state", "00")), + } + + tcpext = _parse_netstat_tcpext() + ip_stats = _parse_snmp_section("Ip") + tcp_stats = _parse_snmp_section("Tcp") + ss_metrics = _get_ss_tcp_metrics() + + retrans_total = tcpext.get("RetransSegs", 0) + ip_in_total = ip_stats.get("InReceives", 0) + ip_out_total = ip_stats.get("OutRequests", 0) + ip_discards_total = ip_stats.get("InDiscards", 0) + ip_stats.get("OutDiscards", 0) + + established = 0 + try: + with open("/proc/net/tcp", "r", encoding="utf-8", errors="ignore") as f: + for line in f.readlines()[1:]: + parts = line.strip().split() + if len(parts) >= 4 and parts[3] == "01": + established += 1 + except OSError: + established = 0 + + prev_ts = NETWORK_STACK_PREV["timestamp"] + dt = max(0.001, now - prev_ts) if prev_ts else 1.0 + + def rate(curr, prev): + if prev is None: + return 0.0 + return max(0.0, (curr - prev) / dt) + + retrans_per_sec = rate(retrans_total, NETWORK_STACK_PREV["tcpext_retrans"]) + ip_in_per_sec = rate(ip_in_total, NETWORK_STACK_PREV["ip_in"]) + ip_out_per_sec = rate(ip_out_total, NETWORK_STACK_PREV["ip_out"]) + ip_drop_per_sec = rate(ip_discards_total, NETWORK_STACK_PREV["ip_discards"]) + + rx_per_sec = 0.0 + tx_per_sec = 0.0 + iface_drop_per_sec = 0.0 + rx_bytes = iface_stats.bytes_recv if iface_stats else 0 + tx_bytes = iface_stats.bytes_sent if iface_stats else 0 + iface_drops = (iface_stats.dropin + iface_stats.dropout) if iface_stats else 0 + if NETWORK_STACK_PREV["iface_rx"] is not None: + rx_per_sec = max(0.0, (rx_bytes - NETWORK_STACK_PREV["iface_rx"]) / dt) + if NETWORK_STACK_PREV["iface_tx"] is not None: + tx_per_sec = max(0.0, (tx_bytes - NETWORK_STACK_PREV["iface_tx"]) / dt) + if NETWORK_STACK_PREV["iface_drops"] is not None: + iface_drop_per_sec = max(0.0, (iface_drops - NETWORK_STACK_PREV["iface_drops"]) / dt) + + NETWORK_STACK_PREV["timestamp"] = now + NETWORK_STACK_PREV["tcpext_retrans"] = retrans_total + NETWORK_STACK_PREV["ip_in"] = ip_in_total + NETWORK_STACK_PREV["ip_out"] = ip_out_total + NETWORK_STACK_PREV["ip_discards"] = ip_discards_total + NETWORK_STACK_PREV["iface_rx"] = rx_bytes + NETWORK_STACK_PREV["iface_tx"] = tx_bytes + NETWORK_STACK_PREV["iface_drops"] = iface_drops + + packets_per_sec = ip_in_per_sec + ip_out_per_sec + drop_ratio = (ip_drop_per_sec / packets_per_sec) if packets_per_sec > 0 else 0.0 + throughput_mb_s = (rx_per_sec + tx_per_sec) / (1024 * 1024) + + retrans_prob = min(0.75, retrans_per_sec / 600.0) + drop_prob = min(0.75, (ip_drop_per_sec / 500.0) + (drop_ratio * 8.0)) + packet_speed = max(1.4, min(4.8, 1.8 + throughput_mb_s / 8.0)) + + socket_activity = min(1.0, (len(all_connections) / 80.0) + (retrans_per_sec / 250.0)) + tcp_activity = min(1.0, (ss_metrics.get("cwnd", 0) / 80.0) + (retrans_per_sec / 300.0)) + ip_activity = min(1.0, packets_per_sec / 15000.0) + netfilter_activity = min(1.0, (ip_drop_per_sec / 120.0) + (drop_ratio * 6.0)) + driver_activity = min(1.0, ((rx_per_sec + tx_per_sec) / (60 * 1024 * 1024)) + (iface_drop_per_sec / 40.0)) + nic_activity = min(1.0, ((rx_per_sec + tx_per_sec) / (80 * 1024 * 1024))) + + return { + "timestamp": datetime.now().isoformat(), + "flow": flow, + "layer_metrics": { + "userspace": {"active_processes": len(psutil.pids())}, + "socket_api": {"active_sockets": len(all_connections), "established": established, "retransmits_per_sec": round(retrans_per_sec, 2)}, + "tcp_udp": { + "established": established, + "retrans_per_sec": round(retrans_per_sec, 2), + "cwnd": int(ss_metrics.get("cwnd", 0)), + "rtt_ms": round(float(ss_metrics.get("rtt_ms", 0.0)), 2), + "tx_queue": int(ss_metrics.get("tx_queue", 0)), + }, + "ip": { + "in_packets_per_sec": round(ip_in_per_sec, 2), + "out_packets_per_sec": round(ip_out_per_sec, 2), + "drop_per_sec": round(ip_drop_per_sec, 3), + "drop_ratio": round(drop_ratio, 5), + }, + "netfilter": {"drop_per_sec": round(ip_drop_per_sec, 3), "drop_ratio": round(drop_ratio, 5)}, + "driver": { + "iface": iface, + "rx_mb_s": round(rx_per_sec / (1024 * 1024), 3), + "tx_mb_s": round(tx_per_sec / (1024 * 1024), 3), + "tx_queue": int(ss_metrics.get("tx_queue", 0)), + "drops_per_sec": round(iface_drop_per_sec, 3), + }, + "nic": { + "iface": iface, + "rx_errors": int(getattr(iface_stats, "errin", 0)) if iface_stats else 0, + "tx_errors": int(getattr(iface_stats, "errout", 0)) if iface_stats else 0, + "drops_total": int(iface_drops), + }, + }, + "layer_activity": { + "userspace": min(1.0, len(psutil.pids()) / 400.0), + "socket": round(socket_activity, 4), + "tcp": round(tcp_activity, 4), + "ip": round(ip_activity, 4), + "netfilter": round(netfilter_activity, 4), + "driver": round(driver_activity, 4), + "nic": round(nic_activity, 4), + }, + "signals": { + "drop_probability": round(drop_prob, 4), + "retransmit_probability": round(retrans_prob, 4), + "packet_speed": round(packet_speed, 3), + }, + "throughput_mb_s": round(throughput_mb_s, 3), + "tcp_counters": { + "in_segs": int(tcp_stats.get("InSegs", 0)), + "out_segs": int(tcp_stats.get("OutSegs", 0)), + "retrans_segs_total": int(retrans_total), + }, + } + + +def get_route_hint(remote_ip): + ip_cmd = resolve_binary("ip") + if not ip_cmd: + return {"remote_ip": remote_ip, "tool": None, "reached": False, "hop_count": 0, "hops": [], "note": "Path tools unavailable on host"} + try: + result = subprocess.run([ip_cmd, "-o", "route", "get", remote_ip], capture_output=True, text=True, timeout=2, check=False) + line = (result.stdout or "").strip() + if not line: + return {"remote_ip": remote_ip, "tool": "ip-route", "reached": False, "hop_count": 0, "hops": [], "note": "No route information available"} + + via_match = re.search(r"\svia\s(\d{1,3}(?:\.\d{1,3}){3})", line) + dev_match = re.search(r"\sdev\s([A-Za-z0-9_.:-]+)", line) + src_match = re.search(r"\ssrc\s(\d{1,3}(?:\.\d{1,3}){3})", line) + + hops = [] + if via_match: + hops.append({"hop": 1, "target": via_match.group(1), "rtt_ms": None}) + hops.append({"hop": 2, "target": remote_ip, "rtt_ms": None}) + else: + hops.append({"hop": 1, "target": remote_ip, "rtt_ms": None}) + + note_parts = ["Traceroute not installed, showing kernel route hint"] + if dev_match: + note_parts.append(f"dev={dev_match.group(1)}") + if src_match: + note_parts.append(f"src={src_match.group(1)}") + + return {"remote_ip": remote_ip, "tool": "ip-route", "reached": False, "hop_count": len(hops), "hops": hops, "note": ", ".join(note_parts)} + except (subprocess.TimeoutExpired, OSError): + return {"remote_ip": remote_ip, "tool": "ip-route", "reached": False, "hop_count": 0, "hops": [], "note": "Route hint lookup timed out"} + + +def get_traceroute_info(remote_ip, max_hops=8): + try: + target_ip = ipaddress.ip_address(remote_ip) + if target_ip.is_loopback or target_ip.is_unspecified: + return {"remote_ip": remote_ip, "tool": None, "reached": False, "hop_count": 0, "hops": [], "note": "Local address, traceroute skipped"} + except ValueError: + raise ValueError("Invalid IP address") + + now = time.time() + cached = TRACEROUTE_CACHE.get(remote_ip) + if cached and (now - cached["timestamp"]) < TRACEROUTE_CACHE_TTL_SECONDS: + return cached["data"] + + traceroute_cmd = resolve_binary("traceroute") + tracepath_cmd = resolve_binary("tracepath") + cmd = None + tool = None + if traceroute_cmd: + cmd = [traceroute_cmd, "-n", "-m", str(max_hops), "-q", "1", "-w", "1", remote_ip] + tool = "traceroute" + elif tracepath_cmd: + cmd = [tracepath_cmd, "-n", "-m", str(max_hops), remote_ip] + tool = "tracepath" + else: + data = get_route_hint(remote_ip) + TRACEROUTE_CACHE[remote_ip] = {"timestamp": now, "data": data} + return data + + try: + result = subprocess.run(cmd, capture_output=True, text=True, timeout=7, check=False) + output = (result.stdout or "").strip() + if not output and result.stderr: + output = result.stderr.strip() + except subprocess.TimeoutExpired: + return {"remote_ip": remote_ip, "tool": tool, "reached": False, "hop_count": 0, "hops": [], "note": "Traceroute timed out"} + + hops = [] + for raw_line in output.splitlines(): + line = raw_line.strip() + hop_match = re.match(r"^(\d+)\s+", line) + if not hop_match: + tracepath_match = re.match(r"^(\d+):\s+", line) + if not tracepath_match: + continue + hop_idx = int(tracepath_match.group(1)) + else: + hop_idx = int(hop_match.group(1)) + + if "*" in line and re.search(r"\*\s*\*\s*\*", line): + hops.append({"hop": hop_idx, "target": "*", "rtt_ms": None}) + continue + + ip_match = re.search(r"(\d{1,3}(?:\.\d{1,3}){3})", line) + rtt_match = re.search(r"(\d+(?:\.\d+)?)\s*ms", line) + hops.append({"hop": hop_idx, "target": ip_match.group(1) if ip_match else "?", "rtt_ms": float(rtt_match.group(1)) if rtt_match else None}) + + reached = any(h.get("target") == remote_ip for h in hops) + data = {"remote_ip": remote_ip, "tool": tool, "reached": reached, "hop_count": len(hops), "hops": hops[:max_hops], "note": None} + TRACEROUTE_CACHE[remote_ip] = {"timestamp": now, "data": data} + return data diff --git a/kernel_ai/services/processes.py b/kernel_ai/services/processes.py new file mode 100644 index 0000000..9cde6ba --- /dev/null +++ b/kernel_ai/services/processes.py @@ -0,0 +1,543 @@ +"""Processes domain service extracted from ``webapp``. + +Pure data builders live here; Flask handlers should only serialize responses. +""" + +from __future__ import annotations + +import math +import os +from datetime import datetime + +import psutil + + +def get_proc_matrix_data() -> list[dict]: + """Build Matrix view data (processes and resource usage).""" + processes = [] + for proc in psutil.process_iter( + ["pid", "name", "cpu_percent", "memory_info", "io_counters", "num_fds"] + ): + try: + info = proc.info + pid = info["pid"] + + cpu_percent = info.get("cpu_percent") or 0.0 + + mem_mb = 0.0 + mem_info = info.get("memory_info") + if mem_info: + mem_mb = mem_info.rss / 1024 / 1024 + + io_total_mb = 0.0 + io_counters = info.get("io_counters") + if io_counters: + io_total_mb = (io_counters.read_bytes + io_counters.write_bytes) / 1024 / 1024 + + net_connections = 0 + tcp_path = f"/proc/{pid}/net/tcp" + try: + if os.path.exists(tcp_path): + with open(tcp_path, "r", encoding="utf-8", errors="ignore") as f: + lines = f.readlines() + net_connections = max(0, len(lines) - 1) + except (IOError, PermissionError): + pass + + num_fds = info.get("num_fds") or 0 + + processes.append( + { + "pid": pid, + "name": info.get("name") or "unknown", + "cpu": float(cpu_percent), + "mem": float(mem_mb), + "io": float(io_total_mb), + "net": int(net_connections), + "fd": int(num_fds), + } + ) + except (psutil.NoSuchProcess, psutil.AccessDenied): + continue + + processes.sort(key=lambda p: p["cpu"], reverse=True) + return processes[:20] + + +def get_processes_detailed_data() -> list[dict]: + """Collect detailed process list for process visualization UI.""" + processes = [] + fields = ["pid", "name", "status", "memory_info", "cpu_percent", "num_threads", "num_fds"] + for proc in psutil.process_iter(fields): + try: + memory_info = proc.info.get("memory_info") + if memory_info is None: + continue + memory_mb = float(memory_info.rss) / 1024 / 1024 + + num_fds = proc.info.get("num_fds") + if num_fds is None or num_fds == 0: + try: + pid = proc.info["pid"] + fd_dir = f"/proc/{pid}/fd" + if os.path.exists(fd_dir): + num_fds = len([f for f in os.listdir(fd_dir) if f.isdigit()]) + else: + num_fds = 0 + except (OSError, PermissionError): + num_fds = 0 + if num_fds is None: + num_fds = 0 + + process_name = proc.info.get("name") or f'pid-{proc.info.get("pid", "unknown")}' + try: + cmdline = proc.cmdline() + if cmdline and len(cmdline) > 0: + if cmdline[0] == "nginx:" and len(cmdline) > 1: + process_name = f"nginx: {cmdline[1]}" + except (psutil.AccessDenied, psutil.NoSuchProcess): + pass + + cmdline_str = "" + try: + cmdline = proc.cmdline() + if cmdline: + cmdline_str = " ".join(cmdline) + except (psutil.AccessDenied, psutil.NoSuchProcess): + pass + + processes.append( + { + "pid": proc.info["pid"], + "name": process_name, + "cmdline": cmdline_str, + "status": proc.info.get("status", "unknown"), + "memory_mb": round(memory_mb, 1), + "cpu_percent": round(float(proc.info.get("cpu_percent", 0) or 0), 1), + "num_threads": int(proc.info.get("num_threads", 0) or 0), + "num_fds": int(num_fds or 0), + } + ) + except (psutil.NoSuchProcess, psutil.AccessDenied, psutil.ZombieProcess): + continue + except Exception: + continue + return processes + + +def _parse_meminfo_kb(): + out = {} + try: + with open("/proc/meminfo", "r", encoding="utf-8", errors="ignore") as f: + for line in f: + parts = line.split() + if len(parts) >= 2 and parts[1].isdigit(): + out[parts[0].rstrip(":")] = int(parts[1]) + except Exception: + pass + return out + + +def _memory_strip_blocks(kind, kb_k, mem_total_kb, n_blocks, seed0): + mem_total_kb = max(1, int(mem_total_kb)) + kb_k = max(0, int(kb_k)) + weights = [] + for i in range(n_blocks): + v = (((seed0 + i * 104729) % 1000) + 40) / 1040.0 + weights.append(v) + sw = sum(weights) + share = kb_k / float(mem_total_kb) + blocks = [] + for i in range(n_blocks): + w = weights[i] / sw + heat = min(1.0, 0.05 + min(0.92, share * 2.0) + (((seed0 + i * 31) % 15) / 120.0)) + blocks.append({"w": round(w, 6), "heat": round(heat, 4), "kind": kind}) + return blocks + + +def _build_memory_visual_rows(meminfo_kb, syscall_nodes, vm, swap): + mi = meminfo_kb or {} + mt = max(1, int(mi.get("MemTotal") or 0)) + if mt <= 1: + try: + mt = max(1, int(getattr(vm, "total", 0) / 1024)) + except Exception: + mt = 1 + + seed_base = (mt % 100000) + int(mi.get("Active", 0) or 0) % 50000 + row_specs = [ + ("buffers", "buffers", mi.get("Buffers", 0)), + ("cached", "page cache", mi.get("Cached", 0)), + ("anon", "anonymous (heap/stack)", mi.get("AnonPages", 0)), + ] + if mi.get("SReclaimable") is not None and mi.get("SUnreclaim") is not None: + row_specs.append(("sreclaim", "slab reclaimable", mi.get("SReclaimable", 0))) + row_specs.append(("sunreclaim", "slab unreclaimable", mi.get("SUnreclaim", 0))) + else: + row_specs.append(("slab", "slab / kmalloc", mi.get("Slab", 0))) + row_specs.extend([("shmem", "shmem / tmpfs", mi.get("Shmem", 0)), ("mapped", "file mappings", mi.get("Mapped", 0))]) + dirty_wb = int(mi.get("Dirty", 0) or 0) + int(mi.get("Writeback", 0) or 0) + int(mi.get("WritebackTmp", 0) or 0) + if dirty_wb > 0: + row_specs.append(("dirty_wb", "dirty + writeback", dirty_wb)) + ah = int(mi.get("AnonHugePages", 0) or 0) + if ah > 0: + row_specs.append(("anon_huge", "transparent huge pages (anon)", ah)) + shm_h = int(mi.get("ShmemHugePages", 0) or 0) + if shm_h > 0: + row_specs.append(("shmem_huge", "huge pages (shmem)", shm_h)) + vmu = int(mi.get("VmallocUsed", 0) or 0) + if vmu > 0: + row_specs.append(("vmalloc", "vmalloc used", vmu)) + ac = int(mi.get("Active", 0) or 0) + iac = int(mi.get("Inactive", 0) or 0) + if ac > 0: + row_specs.append(("active", "active (LRU)", ac)) + if iac > 0: + row_specs.append(("inactive", "inactive (LRU)", iac)) + swap_tot = int(mi.get("SwapTotal", 0) or 0) + swap_free = int(mi.get("SwapFree", 0) or 0) + swap_used = max(0, swap_tot - swap_free) + if swap_tot > 0: + row_specs.append(("swap", "swap occupied", swap_used)) + + pt = int(mi.get("PageTables", 0) or 0) + ks = int(mi.get("KernelStack", 0) or 0) + if pt + ks > 0: + row_specs.append(("kmeta", "pagetables + kernel stacks", pt + ks)) + + rows = [] + for sk, label, kb in row_specs: + kb = int(kb or 0) + if kb <= 0 and sk != "swap": + continue + if sk == "swap" and kb <= 0: + continue + sk_seed = sum(ord(c) for c in sk) * 31 + len(sk) + n_blocks = 22 + (seed_base % 11) + (sk_seed % 9) + seed0 = seed_base + (sk_seed % 100000) + blocks = _memory_strip_blocks(sk, kb, mt, n_blocks, seed0) + rows.append({"id": sk, "label": label, "kb": kb, "pct_of_ram": round(100.0 * kb / float(mt), 2) if mt else 0.0, "blocks": blocks}) + + top_tasks = sorted(syscall_nodes, key=lambda x: int(x.get("rss_bytes") or 0), reverse=True)[:6] + if top_tasks: + tr_bytes = sum(int(x.get("rss_bytes") or 0) for x in top_tasks) or 1 + tr_kb = max(1, int(tr_bytes / 1024)) + task_blocks = [] + for p in top_tasks: + rss = int(p.get("rss_bytes") or 0) + if rss <= 0: + continue + w = rss / float(tr_bytes) + mp = float(p.get("memory_percent") or 0.0) + heat = min(1.0, 0.2 + (mp / 100.0) * 0.75 + (rss / float(tr_bytes)) * 0.15) + task_blocks.append({"w": round(w, 6), "heat": round(heat, 4), "kind": "task", "pid": int(p.get("pid") or 0), "name": str(p.get("name") or "")[:14]}) + if task_blocks: + sw = sum(b["w"] for b in task_blocks) + if sw > 0: + for b in task_blocks: + b["w"] = round(b["w"] / sw, 6) + rows.append({"id": "tasks", "label": "sampled tasks RSS (top)", "kb": tr_kb, "pct_of_ram": round(100.0 * tr_kb / float(mt), 2) if mt else 0.0, "blocks": task_blocks}) + + sr_kb = int(mi.get("SReclaimable", 0) or 0) + su_kb = int(mi.get("SUnreclaim", 0) or 0) + slab_total_kb = int(mi.get("Slab", 0) or 0) or (sr_kb + su_kb) + summary = { + "total_mb": round(mt / 1024.0, 1), + "used_percent": round(vm.percent, 1) if vm else 0.0, + "available_mb": round((mi.get("MemAvailable", 0) or 0) / 1024.0, 1), + "swap_percent": round(swap.percent, 1) if swap else 0.0, + "buffers_mb": round((mi.get("Buffers", 0) or 0) / 1024.0, 1), + "cached_mb": round((mi.get("Cached", 0) or 0) / 1024.0, 1), + "anon_mb": round((mi.get("AnonPages", 0) or 0) / 1024.0, 1), + "slab_mb": round(slab_total_kb / 1024.0, 1), + "sreclaimable_mb": round(sr_kb / 1024.0, 1), + "sunreclaim_mb": round(su_kb / 1024.0, 1), + "dirty_mb": round(int(mi.get("Dirty", 0) or 0) / 1024.0, 2), + "writeback_mb": round(int(mi.get("Writeback", 0) or 0) / 1024.0, 2), + "dirty_writeback_mb": round(dirty_wb / 1024.0, 2), + "anon_huge_mb": round(ah / 1024.0, 2), + "shmem_huge_mb": round(shm_h / 1024.0, 2), + "vmalloc_mb": round(vmu / 1024.0, 2), + "active_mb": round(ac / 1024.0, 1), + "inactive_mb": round(iac / 1024.0, 1), + "swap_used_mb": round(swap_used / 1024.0, 1) if swap_tot else 0.0, + "source": "proc_meminfo+psutil+v2", + } + return rows, summary + + +def collect_processes_realtime(): + """Collect process subsystem telemetry payload.""" + lsm_raw = "" + try: + with open("/sys/kernel/security/lsm", "r", encoding="utf-8", errors="ignore") as f: + lsm_raw = str(f.read().strip()) + except Exception: + lsm_raw = "" + active_lsms = [x.strip() for x in lsm_raw.split(",") if x.strip()] + + yama_scope = "" + try: + with open("/proc/sys/kernel/yama/ptrace_scope", "r", encoding="utf-8", errors="ignore") as f: + yama_scope = str(f.read().strip()) + except Exception: + yama_scope = "" + + syscall_nodes = [] + seccomp_modes = {"none": 0, "strict": 0, "filter": 0, "unknown": 0} + for proc in psutil.process_iter(["pid", "ppid", "name", "username", "cpu_percent", "memory_percent", "num_threads"]): + try: + pid = int(proc.info.get("pid") or 0) + if pid <= 0: + continue + ppid = int(proc.info.get("ppid") or 0) + name = str(proc.info.get("name") or "unknown") + user = str(proc.info.get("username") or "") + cpu = float(proc.info.get("cpu_percent") or 0.0) + mem = float(proc.info.get("memory_percent") or 0.0) + threads = int(proc.info.get("num_threads") or 0) + try: + rss = int(getattr(proc.memory_info(), "rss", 0) or 0) + except Exception: + rss = 0 + try: + fd_count = int(proc.num_fds() or 0) + except Exception: + fd_count = 0 + seccomp_mode = "unknown" + with open(f"/proc/{pid}/status", "r", encoding="utf-8", errors="ignore") as f: + for ln in f: + if ln.startswith("Seccomp:"): + raw = ln.split(":", 1)[1].strip() + if raw == "0": + seccomp_mode = "none" + elif raw == "1": + seccomp_mode = "strict" + elif raw == "2": + seccomp_mode = "filter" + else: + seccomp_mode = "unknown" + break + seccomp_modes[seccomp_mode] = seccomp_modes.get(seccomp_mode, 0) + 1 + syscall_pressure = min(100, int(cpu * 1.5 + threads * 0.35 + mem * 0.8)) + syscall_nodes.append( + { + "pid": pid, + "ppid": ppid, + "name": name, + "user": user, + "fd_count": fd_count, + "syscall_pressure": syscall_pressure, + "seccomp_mode": seccomp_mode, + "memory_percent": round(mem, 2), + "rss_bytes": rss, + } + ) + except Exception: + continue + syscall_nodes.sort(key=lambda x: x.get("syscall_pressure", 0), reverse=True) + syscall_nodes = syscall_nodes[:14] + + network_nodes = {} + try: + for conn in psutil.net_connections(kind="inet"): + pid = int(getattr(conn, "pid", 0) or 0) + if pid <= 0: + continue + remote_ip = "" + try: + raddr = getattr(conn, "raddr", None) + if raddr and len(raddr) >= 1: + remote_ip = str(raddr[0]) + except Exception: + remote_ip = "" + status = str(getattr(conn, "status", "") or "").upper() + bucket = network_nodes.get(pid) + if not bucket: + proc_name = "unknown" + try: + proc_name = psutil.Process(pid).name() + except Exception: + proc_name = "unknown" + bucket = {"pid": pid, "name": proc_name, "connections": 0, "remote_ips": set(), "states": {}} + network_nodes[pid] = bucket + bucket["connections"] += 1 + if remote_ip: + bucket["remote_ips"].add(remote_ip) + if status: + bucket["states"][status] = bucket["states"].get(status, 0) + 1 + except Exception: + pass + + network_tracing = [] + for _, row in network_nodes.items(): + states_sorted = sorted(row["states"].items(), key=lambda kv: kv[1], reverse=True) + top_state = states_sorted[0][0] if states_sorted else "UNKNOWN" + network_tracing.append( + { + "pid": int(row["pid"]), + "name": str(row["name"]), + "connections": int(row["connections"]), + "unique_peers": int(len(row["remote_ips"])), + "peer_sample": sorted(list(row["remote_ips"]))[:4], + "top_state": top_state, + } + ) + network_tracing.sort(key=lambda x: (x.get("connections", 0), x.get("unique_peers", 0)), reverse=True) + network_tracing = network_tracing[:14] + + security_hooks = [ + {"name": "LSM stack", "status": "active" if active_lsms else "unknown", "detail": ",".join(active_lsms[:4]) if active_lsms else "n/a"}, + {"name": "SELinux/AppArmor engines", "status": "active" if any(x in {"selinux", "apparmor"} for x in active_lsms) else "inactive", "detail": "policy-enforcement-path"}, + {"name": "BPF LSM", "status": "active" if "bpf" in active_lsms else "inactive", "detail": "dynamic-policy-hook"}, + {"name": "seccomp filter gate", "status": "active" if (seccomp_modes.get("filter", 0) + seccomp_modes.get("strict", 0)) > 0 else "inactive", "detail": f"filter:{seccomp_modes.get('filter', 0)} strict:{seccomp_modes.get('strict', 0)}"}, + {"name": "Yama ptrace scope", "status": "hardened" if yama_scope in {"2", "3"} else ("relaxed" if yama_scope in {"0", "1"} else "unknown"), "detail": yama_scope or "n/a"}, + ] + + node_pool = {} + for row in syscall_nodes[:16]: + pid = int(row.get("pid") or 0) + if pid <= 0: + continue + node_pool[pid] = { + "pid": pid, + "ppid": int(row.get("ppid") or 0), + "name": str(row.get("name") or "unknown"), + "user": str(row.get("user") or ""), + "syscall_pressure": int(row.get("syscall_pressure") or 0), + "fd_count": int(row.get("fd_count") or 0), + "seccomp_mode": str(row.get("seccomp_mode") or "unknown"), + "connections": 0, + "unique_peers": 0, + "memory_percent": float(row.get("memory_percent") or 0.0), + "rss_bytes": int(row.get("rss_bytes") or 0), + } + for row in network_tracing[:16]: + pid = int(row.get("pid") or 0) + if pid <= 0: + continue + if pid not in node_pool: + node_pool[pid] = { + "pid": pid, + "ppid": 0, + "name": str(row.get("name") or "unknown"), + "user": "", + "syscall_pressure": 0, + "fd_count": 0, + "seccomp_mode": "unknown", + "connections": 0, + "unique_peers": 0, + "memory_percent": 0.0, + "rss_bytes": 0, + } + node_pool[pid]["connections"] = int(row.get("connections") or 0) + node_pool[pid]["unique_peers"] = int(row.get("unique_peers") or 0) + + edges = [] + edge_keys = set() + network_by_pid = {int(r.get("pid") or 0): r for r in network_tracing} + node_pids = sorted(node_pool.keys()) + + def _add_edge(src_pid, dst_pid, edge_type, weight): + src = int(src_pid or 0) + dst = int(dst_pid or 0) + if src <= 0 or dst <= 0 or src == dst: + return + if src not in node_pool or dst not in node_pool: + return + pair = tuple(sorted((src, dst))) + key = (pair[0], pair[1], edge_type) + if key in edge_keys: + return + edge_keys.add(key) + edges.append({"source": src, "target": dst, "type": edge_type, "weight": float(max(0.1, min(1.0, weight)))}) + + for pid, node in node_pool.items(): + ppid = int(node.get("ppid") or 0) + if ppid in node_pool: + _add_edge(pid, ppid, "ipc", 0.72) + + sorted_by_pressure = sorted(node_pool.values(), key=lambda n: n.get("syscall_pressure", 0), reverse=True) + for i in range(len(sorted_by_pressure) - 1): + a = sorted_by_pressure[i] + b = sorted_by_pressure[i + 1] + diff = abs(int(a.get("syscall_pressure", 0)) - int(b.get("syscall_pressure", 0))) + weight = 1.0 - min(0.8, diff / 100.0) + _add_edge(int(a.get("pid")), int(b.get("pid")), "syscalls", weight) + + for i in range(len(node_pids)): + for j in range(i + 1, len(node_pids)): + pa = node_pids[i] + pb = node_pids[j] + ra = network_by_pid.get(pa) or {} + rb = network_by_pid.get(pb) or {} + sa = set(ra.get("peer_sample") or []) + sb = set(rb.get("peer_sample") or []) + if sa and sb and (sa & sb): + _add_edge(pa, pb, "network", 0.88) + + for i in range(len(node_pids)): + for j in range(i + 1, len(node_pids)): + na = node_pool[node_pids[i]] + nb = node_pool[node_pids[j]] + if not na.get("user") or na.get("user") != nb.get("user"): + continue + fa = int(na.get("fd_count") or 0) + fb = int(nb.get("fd_count") or 0) + if fa >= 16 and fb >= 16: + _add_edge(int(na.get("pid")), int(nb.get("pid")), "file_access", 0.64) + + nodes = list(node_pool.values())[:18] + edges = edges[:64] + + try: + vm = psutil.virtual_memory() + swap = psutil.swap_memory() + meminfo_kb = _parse_meminfo_kb() + strip_rows, mem_summary = _build_memory_visual_rows(meminfo_kb, syscall_nodes, vm, swap) + memory_visual = {"layout": "strips", "rows": strip_rows, "summary": mem_summary} + except Exception: + memory_visual = { + "layout": "strips", + "rows": [], + "summary": {"total_mb": 0, "used_percent": 0.0, "available_mb": 0, "swap_percent": 0.0, "source": "error"}, + } + + return { + "timestamp": datetime.utcnow().isoformat() + "Z", + "syscalls_interception": syscall_nodes, + "network_tracing": network_tracing, + "security_hooks": security_hooks, + "neural_graph": {"nodes": nodes, "edges": edges}, + "memory_visual": memory_visual, + "meta": { + "processes_sampled": len(syscall_nodes), + "network_processes": len(network_tracing), + "seccomp_filter_percent": round((seccomp_modes.get("filter", 0) + seccomp_modes.get("strict", 0)) * 100.0 / max(1, sum(seccomp_modes.values())), 2), + "mode": "live-heuristic-v1", + }, + } + + +def get_proc_graph_data(): + """Data for ``/api/proc-graph``.""" + matrix = get_proc_matrix_data() + nodes = [{"id": "kernel", "type": "kernel", "label": "kernel", "x": 0.0, "y": 0.0, "z": 0.0}] + edges = [] + for i, p in enumerate(matrix[:16]): + angle = (2 * math.pi * i) / max(len(matrix[:16]), 1) + r = 75.0 + pid = p.get("pid", i) + nid = f"proc-{pid}" + nodes.append( + { + "id": nid, + "type": "process", + "label": str(p.get("name", "process")), + "x": r * math.cos(angle), + "y": r * math.sin(angle), + "z": 18.0 * math.sin(i * 0.45), + } + ) + edges.append({"from": nid, "to": "kernel"}) + return {"nodes": nodes, "edges": edges, "timestamp": datetime.now().isoformat()} diff --git a/kernel_ai/services/system_view.py b/kernel_ai/services/system_view.py new file mode 100644 index 0000000..d8ecf09 --- /dev/null +++ b/kernel_ai/services/system_view.py @@ -0,0 +1,321 @@ +"""Filesystem and isolation domain services.""" + +from __future__ import annotations + +import os +import re +import time +from datetime import datetime + +import psutil + +from kernel_ai.collectors import proc_fs as _proc_fs +from kernel_ai.state import FILESYSTEM_PREV + + +def get_filesystem_blocks(): + now = time.time() + try: + usage = psutil.disk_usage("/") + except Exception: + usage = None + + used_percent = float(usage.percent) if usage else 0.0 + total_gb = round((usage.total / (1024**3)), 2) if usage else 0.0 + used_gb = round((usage.used / (1024**3)), 2) if usage else 0.0 + free_gb = round((usage.free / (1024**3)), 2) if usage else 0.0 + + io = psutil.disk_io_counters() + write_bytes = int(io.write_bytes) if io else 0 + prev_ts = FILESYSTEM_PREV["timestamp"] + prev_write = FILESYSTEM_PREV["write_bytes"] + dt = max(0.001, now - prev_ts) if prev_ts else 1.0 + write_bps = 0.0 if prev_write is None else max(0.0, (write_bytes - prev_write) / dt) + + rows = 20 + cols = 34 + total_blocks = rows * cols + used_ratio_global = max(0.0, min(1.0, used_percent / 100.0)) + + zone_defs = [ + {"id": "root", "name": "/", "path": "/", "base": 1.5, "bias": 0.00}, + {"id": "var", "name": "/var", "path": "/var", "base": 1.6, "bias": 0.10}, + {"id": "home", "name": "/home", "path": "/home", "base": 1.35, "bias": 0.06}, + {"id": "usr", "name": "/usr", "path": "/usr", "base": 1.45, "bias": 0.08}, + {"id": "etc", "name": "/etc", "path": "/etc", "base": 1.0, "bias": -0.03}, + {"id": "tmp", "name": "/tmp", "path": "/tmp", "base": 1.0, "bias": -0.02}, + {"id": "dev", "name": "/dev", "path": "/dev", "base": 0.85, "bias": -0.06}, + ] + + activity_counts = {z["id"]: 0 for z in zone_defs} + try: + processes = list(psutil.process_iter(["pid"]))[:90] + zone_paths = sorted([(z["path"], z["id"]) for z in zone_defs], key=lambda x: len(x[0]), reverse=True) + for proc in processes: + try: + open_files = proc.open_files()[:28] + except (psutil.AccessDenied, psutil.NoSuchProcess, psutil.ZombieProcess, OSError): + continue + for of in open_files: + fpath = str(getattr(of, "path", "") or "") + if not fpath.startswith("/"): + continue + for prefix, zone_id in zone_paths: + if prefix == "/": + continue + if fpath == prefix or fpath.startswith(prefix + "/"): + activity_counts[zone_id] += 1 + break + else: + activity_counts["root"] += 1 + except Exception: + pass + + weighted = [] + for z in zone_defs: + act = float(activity_counts.get(z["id"], 0)) + z["activity"] = act + weighted.append(max(0.2, z["base"] + act * 0.08)) + + total_weight = sum(weighted) or 1.0 + row_counts = [max(1, int(round(rows * w / total_weight))) for w in weighted] + while sum(row_counts) > rows: + idx = max(range(len(row_counts)), key=lambda i: row_counts[i]) + if row_counts[idx] > 1: + row_counts[idx] -= 1 + else: + break + while sum(row_counts) < rows: + idx = max(range(len(weighted)), key=lambda i: weighted[i]) + row_counts[idx] += 1 + + writing_ratio = min(0.20, write_bps / (300 * 1024 * 1024)) + writing_blocks_total = int(round(total_blocks * writing_ratio)) + writing_blocks_total = max(0, min(total_blocks, writing_blocks_total)) + + blocks = [] + zones = [] + cursor_row = 0 + zone_scores = [] + for z in zone_defs: + zone_scores.append(z["activity"] + 1.0) + score_sum = sum(zone_scores) or 1.0 + + seed = int(now * 3) + for idx, z in enumerate(zone_defs): + row_span = row_counts[idx] + row_start = cursor_row + row_end = min(rows - 1, cursor_row + row_span - 1) + cursor_row += row_span + + zone_cells = max(1, (row_end - row_start + 1) * cols) + local_used_ratio = max(0.05, min(0.98, used_ratio_global + z["bias"] + min(0.18, z["activity"] / 120.0))) + zone_used = int(round(zone_cells * local_used_ratio)) + zone_used = max(0, min(zone_cells, zone_used)) + + zone_write_share = zone_scores[idx] / score_sum + zone_writing = int(round(writing_blocks_total * zone_write_share)) + zone_writing = max(0, min(zone_used, zone_writing)) + inode_pressure = int(max(0, min(100, round((z["activity"] * 2.6) + (zone_writing * 0.9) + (local_used_ratio * 38.0))))) + + cell_index = 0 + for r in range(row_start, row_end + 1): + for c in range(cols): + state = "used" if cell_index < zone_used else "free" + blocks.append({"r": r, "c": c, "i": r * cols + c, "zone_id": z["id"], "state": state}) + cell_index += 1 + + if zone_writing > 0 and zone_used > 0: + zone_block_indices = [i for i, b in enumerate(blocks) if b["zone_id"] == z["id"] and b["state"] == "used"] + used_len = len(zone_block_indices) + for n in range(min(zone_writing, used_len)): + pick = (seed * 31 + idx * 67 + n * 43) % used_len + blocks[zone_block_indices[pick]]["state"] = "writing" + + zones.append( + { + "id": z["id"], + "name": z["name"], + "path": z["path"], + "row_start": row_start, + "row_end": row_end, + "activity": int(z["activity"]), + "used_percent": round(local_used_ratio * 100.0, 1), + "writing_blocks": zone_writing, + "inode_pressure": inode_pressure, + } + ) + + writing_blocks = sum(1 for b in blocks if b["state"] == "writing") + FILESYSTEM_PREV["timestamp"] = now + FILESYSTEM_PREV["write_bytes"] = write_bytes + + inode_pressure_global = 0 + if zones: + inode_pressure_global = int(round(sum(int(z.get("inode_pressure", 0)) for z in zones) / len(zones))) + + return { + "timestamp": datetime.now().isoformat(), + "rows": rows, + "cols": cols, + "zones": zones, + "blocks": blocks, + "meta": { + "total_gb": total_gb, + "used_gb": used_gb, + "free_gb": free_gb, + "used_percent": round(used_percent, 2), + "write_bps": round(write_bps, 2), + "writing_blocks": writing_blocks, + "inode_pressure": inode_pressure_global, + }, + } + + +def _parse_cgroup_path(pid): + cgroup_text = _proc_fs.safe_read_text(f"/proc/{pid}/cgroup") + if not cgroup_text: + return "/" + chosen = "/" + for line in cgroup_text.splitlines(): + parts = line.split(":") + if len(parts) != 3: + continue + _, controllers, path = parts + path = path.strip() or "/" + if controllers == "": + return path + if path and path != "/": + chosen = path + return chosen + + +def _read_namespace_inode(pid, ns_name): + ns_link = f"/proc/{pid}/ns/{ns_name}" + try: + target = os.readlink(ns_link) + except (OSError, PermissionError): + return None + match = re.search(r"\[(\d+)\]", target) + return match.group(1) if match else target + + +def _read_cgroup_v2_stats(cgroup_path): + root = "/sys/fs/cgroup" + rel = cgroup_path.lstrip("/") + base = os.path.join(root, rel) if rel else root + + cpu_max_text = _proc_fs.safe_read_text(os.path.join(base, "cpu.max")) + cpu_quota_cores = None + if cpu_max_text: + parts = cpu_max_text.split() + if len(parts) >= 2 and parts[0] != "max": + try: + quota = float(parts[0]) + period = float(parts[1]) + if period > 0: + cpu_quota_cores = round(quota / period, 2) + except ValueError: + cpu_quota_cores = None + + mem_current = _proc_fs.safe_read_text(os.path.join(base, "memory.current")) + mem_max = _proc_fs.safe_read_text(os.path.join(base, "memory.max")) + pids_current = _proc_fs.safe_read_text(os.path.join(base, "pids.current")) + pids_max = _proc_fs.safe_read_text(os.path.join(base, "pids.max")) + io_stat_text = _proc_fs.safe_read_text(os.path.join(base, "io.stat")) + + memory_current_mb = None + memory_max_mb = None + try: + if mem_current is not None: + memory_current_mb = round(int(mem_current) / (1024 * 1024), 1) + except ValueError: + memory_current_mb = None + + try: + if mem_max and mem_max != "max": + memory_max_mb = round(int(mem_max) / (1024 * 1024), 1) + except ValueError: + memory_max_mb = None + + io_bytes = None + if io_stat_text: + total = 0 + for line in io_stat_text.splitlines(): + rbytes_match = re.search(r"rbytes=(\d+)", line) + wbytes_match = re.search(r"wbytes=(\d+)", line) + if rbytes_match: + total += int(rbytes_match.group(1)) + if wbytes_match: + total += int(wbytes_match.group(1)) + io_bytes = total + + return { + "cpu_quota_cores": cpu_quota_cores, + "memory_current_mb": memory_current_mb, + "memory_max_mb": memory_max_mb, + "pids_current": int(pids_current) if pids_current and pids_current.isdigit() else None, + "pids_max": None if pids_max in (None, "max") else (int(pids_max) if pids_max.isdigit() else None), + "io_total_mb": round(io_bytes / (1024 * 1024), 1) if io_bytes is not None else None, + } + + +def get_isolation_context(): + namespace_keys = ["mnt", "pid", "net", "ipc", "uts", "user"] + namespace_labels = {"mnt": "MNT", "pid": "PID", "net": "NET", "ipc": "IPC", "uts": "UTS", "user": "USER"} + namespace_counts = {k: {} for k in namespace_keys} + cgroup_aggregates = {} + total_scanned = 0 + + for proc in psutil.process_iter(["pid", "name", "memory_info"]): + try: + pid = proc.info["pid"] + total_scanned += 1 + + cgroup_path = _parse_cgroup_path(pid) + agg = cgroup_aggregates.setdefault(cgroup_path, {"path": cgroup_path, "process_count": 0, "memory_mb_sum": 0.0, "sample_processes": []}) + agg["process_count"] += 1 + + mem_info = proc.info.get("memory_info") + if mem_info: + agg["memory_mb_sum"] += mem_info.rss / (1024 * 1024) + + if len(agg["sample_processes"]) < 4: + process_name = proc.info.get("name") or "unknown" + agg["sample_processes"].append(process_name) + + for ns_name in namespace_keys: + inode = _read_namespace_inode(pid, ns_name) + if inode: + ns_map = namespace_counts[ns_name] + ns_map[inode] = ns_map.get(inode, 0) + 1 + except (psutil.NoSuchProcess, psutil.AccessDenied, KeyError): + continue + + namespaces = [] + for ns_name in namespace_keys: + entries = namespace_counts[ns_name] + unique_count = len(entries) + dominant_inode = None + dominant_count = 0 + if entries: + dominant_inode, dominant_count = max(entries.items(), key=lambda kv: kv[1]) + activity = round((dominant_count / total_scanned), 3) if total_scanned > 0 else 0 + namespaces.append( + { + "id": ns_name, + "label": namespace_labels[ns_name], + "unique_count": unique_count, + "dominant_inode": dominant_inode, + "dominant_count": dominant_count, + "activity": activity, + } + ) + + top_cgroups = sorted(cgroup_aggregates.values(), key=lambda x: (x["process_count"], x["memory_mb_sum"]), reverse=True)[:4] + for item in top_cgroups: + stats = _read_cgroup_v2_stats(item["path"]) + item["memory_mb_sum"] = round(item["memory_mb_sum"], 1) + item.update(stats) + + return {"timestamp": datetime.now().isoformat(), "processes_scanned": total_scanned, "namespaces": namespaces, "top_cgroups": top_cgroups} diff --git a/kernel_ai/webapp.py b/kernel_ai/webapp.py index 3048aa1..57da5ed 100644 --- a/kernel_ai/webapp.py +++ b/kernel_ai/webapp.py @@ -12,7 +12,6 @@ import platform import subprocess import re -import ipaddress import shutil from datetime import datetime from flask import Flask, jsonify, render_template, send_from_directory, request, redirect @@ -23,18 +22,17 @@ from kernel_ai.http.register import register_http_routes from kernel_ai.prometheus_setup import init_prometheus from kernel_ai.collectors import proc_fs as _proc_fs +from kernel_ai.services import network as _network_service +from kernel_ai.services import processes as _processes_service +from kernel_ai.services import system_view as _system_view_service from kernel_ai.state import ( CRYPTO_PREV, DEVICES_PREV, ENTROPY_PREV, EXEC_CONTEXT_PREV, - FILESYSTEM_PREV, FRONTEND_LOG_FILE, FRONTEND_LOG_WRITE_LOCK, - NETWORK_STACK_PREV, SECURITY_PREV, - TRACEROUTE_CACHE, - TRACEROUTE_CACHE_TTL_SECONDS, ) # Use ``_proc_fs.*`` directly so tests can monkeypatch ``kernel_ai.collectors.proc_fs``. @@ -661,66 +659,7 @@ def get_mock_process_kernel_map(): def get_proc_matrix_data(): """Build Matrix view data - processes and their resource usage""" - matrix = [] - - # Collect processes with required fields - processes = [] - for proc in psutil.process_iter( - ['pid', 'name', 'cpu_percent', 'memory_info', 'io_counters', 'num_fds'] - ): - try: - info = proc.info - pid = info['pid'] - - # CPU usage (may be 0 on first call) - cpu_percent = info.get('cpu_percent') or 0.0 - - # Memory: resident set size in MB - mem_mb = 0.0 - mem_info = info.get('memory_info') - if mem_info: - mem_mb = mem_info.rss / 1024 / 1024 - - # IO: sum of read/write bytes in MB - io_total_mb = 0.0 - io_counters = info.get('io_counters') - if io_counters: - io_total_mb = ( - io_counters.read_bytes + io_counters.write_bytes - ) / 1024 / 1024 - - # NET: count TCP entries from /proc/[pid]/net/tcp - net_connections = 0 - tcp_path = f'/proc/{pid}/net/tcp' - try: - if os.path.exists(tcp_path): - with open(tcp_path, 'r') as f: - lines = f.readlines() - # subtract header - net_connections = max(0, len(lines) - 1) - except (IOError, PermissionError): - pass - - # FD: number of file descriptors - num_fds = info.get('num_fds') or 0 - - processes.append({ - 'pid': pid, - 'name': info.get('name') or 'unknown', - 'cpu': float(cpu_percent), - 'mem': float(mem_mb), - 'io': float(io_total_mb), - 'net': int(net_connections), - 'fd': int(num_fds), - }) - except (psutil.NoSuchProcess, psutil.AccessDenied): - continue - - # Sort by CPU usage and take top 20 for clarity - processes.sort(key=lambda p: p['cpu'], reverse=True) - matrix = processes[:20] - - return matrix + return _processes_service.get_proc_matrix_data() # API Endpoints @@ -907,195 +846,28 @@ def nginx_files(): except Exception as e: return jsonify({"error": str(e)}), 500 def get_active_connections(): - """Get active network connections""" - try: - connections = [] - # Get TCP connections - with open("/proc/net/tcp", "r") as f: - lines = f.readlines()[1:] # Skip header - for line in lines: - parts = line.strip().split() - if len(parts) >= 4: - local_addr = parts[1] - remote_addr = parts[2] - state = parts[3] - - # Convert hex addresses to readable format - # IP addresses in /proc/net/tcp are stored in little-endian format - def hex_to_ip(hex_str): - # Reverse the hex string to convert from little-endian - hex_bytes = [hex_str[i:i+2] for i in range(0, 8, 2)] - hex_bytes.reverse() - return ".".join([str(int(b, 16)) for b in hex_bytes]) - - local_ip = hex_to_ip(local_addr.split(":")[0]) - local_port = int(local_addr.split(":")[1], 16) - - if remote_addr != "00000000:0000": # Not listening - remote_ip = hex_to_ip(remote_addr.split(":")[0]) - remote_port = int(remote_addr.split(":")[1], 16) - - connections.append({ - "local": f"{local_ip}:{local_port}", - "remote": f"{remote_ip}:{remote_port}", - "state": state, - "type": "TCP" - }) - - # Limit to first 20 connections for display - return connections[:20] - - except Exception as e: - print(f"Error getting active connections: {e}") - return get_mock_active_connections() + """Get active network connections.""" + return _network_service.get_active_connections() def get_mock_active_connections(): - """Mock data for active connections""" - return [ - {"local": "127.0.0.1:22", "remote": "192.168.1.100:54321", "state": "01", "type": "TCP"}, - {"local": "0.0.0.0:80", "remote": "10.0.0.50:12345", "state": "01", "type": "TCP"}, - {"local": "127.0.0.1:3306", "remote": "172.16.0.10:65432", "state": "01", "type": "TCP"}, - {"local": "0.0.0.0:443", "remote": "203.0.113.0:54321", "state": "01", "type": "TCP"}, - {"local": "127.0.0.1:5001", "remote": "192.168.1.101:12345", "state": "01", "type": "TCP"} - ] + """Mock data for active connections.""" + return _network_service.get_mock_active_connections() def _tcp_state_name(code): - states = { - "01": "ESTABLISHED", - "02": "SYN_SENT", - "03": "SYN_RECV", - "04": "FIN_WAIT1", - "05": "FIN_WAIT2", - "06": "TIME_WAIT", - "07": "CLOSE", - "08": "CLOSE_WAIT", - "09": "LAST_ACK", - "0A": "LISTEN", - "0B": "CLOSING" - } - return states.get(str(code).upper(), str(code).upper()) + return _network_service._tcp_state_name(code) def _get_default_iface(): - try: - with open("/proc/net/route", "r") as f: - lines = f.readlines()[1:] - for line in lines: - parts = line.strip().split() - if len(parts) < 11: - continue - iface = parts[0] - destination = parts[1] - flags = int(parts[3], 16) - if destination == "00000000" and (flags & 0x2): - return iface - except (OSError, ValueError): - pass - # Fallback: first non-loopback interface. - try: - pernic = psutil.net_io_counters(pernic=True) - for iface in pernic.keys(): - if iface != "lo": - return iface - except Exception: - pass - return "lo" + return _network_service._get_default_iface() def _parse_netstat_tcpext(): - try: - with open("/proc/net/netstat", "r") as f: - lines = [line.strip() for line in f if line.strip()] - for i in range(0, len(lines) - 1, 2): - header = lines[i].split() - values = lines[i + 1].split() - if not header or header[0] != "TcpExt:": - continue - if not values or values[0] != "TcpExt:": - continue - fields = header[1:] - nums = values[1:] - if len(fields) != len(nums): - continue - mapping = {} - for name, val in zip(fields, nums): - try: - mapping[name] = int(val) - except ValueError: - mapping[name] = 0 - return mapping - except OSError: - return {} - return {} + return _network_service._parse_netstat_tcpext() def _parse_snmp_section(section_name): - try: - with open("/proc/net/snmp", "r") as f: - lines = [line.strip() for line in f if line.strip()] - for i in range(0, len(lines) - 1, 2): - header = lines[i].split() - values = lines[i + 1].split() - expected_prefix = f"{section_name}:" - if not header or header[0] != expected_prefix: - continue - if not values or values[0] != expected_prefix: - continue - fields = header[1:] - nums = values[1:] - if len(fields) != len(nums): - continue - out = {} - for name, val in zip(fields, nums): - try: - out[name] = int(val) - except ValueError: - out[name] = 0 - return out - except OSError: - return {} - return {} + return _network_service._parse_snmp_section(section_name) def _get_ss_tcp_metrics(): """Extract cwnd/rtt/retrans and tx queue from ss -tin (best effort).""" - ss_cmd = resolve_binary("ss") - if not ss_cmd: - return {} - try: - result = subprocess.run( - [ss_cmd, "-tin"], - capture_output=True, - text=True, - timeout=2, - check=False - ) - lines = (result.stdout or "").splitlines() - except (subprocess.TimeoutExpired, OSError): - return {} - - for idx, line in enumerate(lines): - if not line.strip().startswith("ESTAB"): - continue - metrics = {} - parts = line.split() - # ESTAB Recv-Q Send-Q Local:Port Peer:Port - if len(parts) >= 4: - try: - metrics["tx_queue"] = int(parts[2]) - metrics["rx_queue"] = int(parts[1]) - except ValueError: - pass - - details = lines[idx + 1] if (idx + 1) < len(lines) else "" - rtt_match = re.search(r'rtt:(\d+(?:\.\d+)?)/', details) - cwnd_match = re.search(r'cwnd:(\d+)', details) - retrans_match = re.search(r'retrans:(\d+)(?:/\d+)?', details) - if rtt_match: - metrics["rtt_ms"] = float(rtt_match.group(1)) - if cwnd_match: - metrics["cwnd"] = int(cwnd_match.group(1)) - if retrans_match: - metrics["retrans_now"] = int(retrans_match.group(1)) - if metrics: - return metrics - return {} + return _network_service._get_ss_tcp_metrics() def _read_major_minor_from_devfile(devfile_path): value = _proc_fs.safe_read_text(devfile_path) @@ -1407,681 +1179,31 @@ def get_devices_realtime(): } def get_filesystem_blocks(): - now = time.time() - try: - usage = psutil.disk_usage("/") - except Exception: - usage = None - - used_percent = float(usage.percent) if usage else 0.0 - total_gb = round((usage.total / (1024 ** 3)), 2) if usage else 0.0 - used_gb = round((usage.used / (1024 ** 3)), 2) if usage else 0.0 - free_gb = round((usage.free / (1024 ** 3)), 2) if usage else 0.0 - - io = psutil.disk_io_counters() - write_bytes = int(io.write_bytes) if io else 0 - prev_ts = FILESYSTEM_PREV["timestamp"] - prev_write = FILESYSTEM_PREV["write_bytes"] - dt = max(0.001, now - prev_ts) if prev_ts else 1.0 - write_bps = 0.0 if prev_write is None else max(0.0, (write_bytes - prev_write) / dt) - - rows = 20 - cols = 34 - total_blocks = rows * cols - used_ratio_global = max(0.0, min(1.0, used_percent / 100.0)) - - # Logical filesystem zones for a visible map layout. - zone_defs = [ - {"id": "root", "name": "/", "path": "/", "base": 1.5, "bias": 0.00}, - {"id": "var", "name": "/var", "path": "/var", "base": 1.6, "bias": 0.10}, - {"id": "home", "name": "/home", "path": "/home", "base": 1.35, "bias": 0.06}, - {"id": "usr", "name": "/usr", "path": "/usr", "base": 1.45, "bias": 0.08}, - {"id": "etc", "name": "/etc", "path": "/etc", "base": 1.0, "bias": -0.03}, - {"id": "tmp", "name": "/tmp", "path": "/tmp", "base": 1.0, "bias": -0.02}, - {"id": "dev", "name": "/dev", "path": "/dev", "base": 0.85, "bias": -0.06}, - ] - - activity_counts = {z["id"]: 0 for z in zone_defs} - - # Best-effort activity sampling from open file descriptors by path prefix. - try: - processes = list(psutil.process_iter(["pid"]))[:90] - zone_paths = sorted([(z["path"], z["id"]) for z in zone_defs], key=lambda x: len(x[0]), reverse=True) - for proc in processes: - try: - open_files = proc.open_files()[:28] - except (psutil.AccessDenied, psutil.NoSuchProcess, psutil.ZombieProcess, OSError): - continue - for of in open_files: - fpath = str(getattr(of, "path", "") or "") - if not fpath.startswith("/"): - continue - for prefix, zone_id in zone_paths: - if prefix == "/": - continue - if fpath == prefix or fpath.startswith(prefix + "/"): - activity_counts[zone_id] += 1 - break - else: - activity_counts["root"] += 1 - except Exception: - pass - - weighted = [] - for z in zone_defs: - act = float(activity_counts.get(z["id"], 0)) - z["activity"] = act - weighted.append(max(0.2, z["base"] + act * 0.08)) - - total_weight = sum(weighted) or 1.0 - row_counts = [max(1, int(round(rows * w / total_weight))) for w in weighted] - # Normalize row counts to exact total rows. - while sum(row_counts) > rows: - idx = max(range(len(row_counts)), key=lambda i: row_counts[i]) - if row_counts[idx] > 1: - row_counts[idx] -= 1 - else: - break - while sum(row_counts) < rows: - idx = max(range(len(weighted)), key=lambda i: weighted[i]) - row_counts[idx] += 1 - - writing_ratio = min(0.20, write_bps / (300 * 1024 * 1024)) - writing_blocks_total = int(round(total_blocks * writing_ratio)) - writing_blocks_total = max(0, min(total_blocks, writing_blocks_total)) - - blocks = [] - zones = [] - cursor_row = 0 - zone_used_total = 0 - zone_writing_total = 0 - zone_scores = [] - for z in zone_defs: - zone_scores.append(z["activity"] + 1.0) - score_sum = sum(zone_scores) or 1.0 - - seed = int(now * 3) - for idx, z in enumerate(zone_defs): - row_span = row_counts[idx] - row_start = cursor_row - row_end = min(rows - 1, cursor_row + row_span - 1) - cursor_row += row_span - - zone_cells = max(1, (row_end - row_start + 1) * cols) - local_used_ratio = max(0.05, min(0.98, used_ratio_global + z["bias"] + min(0.18, z["activity"] / 120.0))) - zone_used = int(round(zone_cells * local_used_ratio)) - zone_used = max(0, min(zone_cells, zone_used)) - zone_used_total += zone_used - - zone_write_share = zone_scores[idx] / score_sum - zone_writing = int(round(writing_blocks_total * zone_write_share)) - zone_writing = max(0, min(zone_used, zone_writing)) - zone_writing_total += zone_writing - inode_pressure = int(max(0, min( - 100, - round((z["activity"] * 2.6) + (zone_writing * 0.9) + (local_used_ratio * 38.0)) - ))) - - cell_index = 0 - for r in range(row_start, row_end + 1): - for c in range(cols): - state = "used" if cell_index < zone_used else "free" - blocks.append({ - "r": r, - "c": c, - "i": r * cols + c, - "zone_id": z["id"], - "state": state - }) - cell_index += 1 - - if zone_writing > 0 and zone_used > 0: - # Convert some used cells into writing cells within this zone segment. - zone_block_indices = [ - i for i, b in enumerate(blocks) - if b["zone_id"] == z["id"] and b["state"] == "used" - ] - used_len = len(zone_block_indices) - for n in range(min(zone_writing, used_len)): - pick = (seed * 31 + idx * 67 + n * 43) % used_len - blocks[zone_block_indices[pick]]["state"] = "writing" - - zones.append({ - "id": z["id"], - "name": z["name"], - "path": z["path"], - "row_start": row_start, - "row_end": row_end, - "activity": int(z["activity"]), - "used_percent": round(local_used_ratio * 100.0, 1), - "writing_blocks": zone_writing, - "inode_pressure": inode_pressure - }) - - writing_blocks = sum(1 for b in blocks if b["state"] == "writing") - - FILESYSTEM_PREV["timestamp"] = now - FILESYSTEM_PREV["write_bytes"] = write_bytes - - inode_pressure_global = 0 - if zones: - inode_pressure_global = int(round(sum(int(z.get("inode_pressure", 0)) for z in zones) / len(zones))) - - return { - "timestamp": datetime.now().isoformat(), - "rows": rows, - "cols": cols, - "zones": zones, - "blocks": blocks, - "meta": { - "total_gb": total_gb, - "used_gb": used_gb, - "free_gb": free_gb, - "used_percent": round(used_percent, 2), - "write_bps": round(write_bps, 2), - "writing_blocks": writing_blocks, - "inode_pressure": inode_pressure_global - } - } + return _system_view_service.get_filesystem_blocks() def get_network_stack_realtime(): - now = time.time() - iface = _get_default_iface() - pernic = psutil.net_io_counters(pernic=True) - iface_stats = pernic.get(iface) - all_connections = get_active_connections() - interesting = [ - c for c in all_connections - if not c["remote"].startswith("127.0.0.1") and not c["remote"].startswith("0.0.0.0") - ] - flow = interesting[0] if interesting else (all_connections[0] if all_connections else None) - if flow: - flow = { - "local": flow.get("local"), - "remote": flow.get("remote"), - "type": str(flow.get("type", "TCP")).upper(), - "state_code": flow.get("state", "00"), - "state_name": _tcp_state_name(flow.get("state", "00")) - } - - tcpext = _parse_netstat_tcpext() - ip_stats = _parse_snmp_section("Ip") - tcp_stats = _parse_snmp_section("Tcp") - ss_metrics = _get_ss_tcp_metrics() - - retrans_total = tcpext.get("RetransSegs", 0) - ip_in_total = ip_stats.get("InReceives", 0) - ip_out_total = ip_stats.get("OutRequests", 0) - ip_discards_total = ip_stats.get("InDiscards", 0) + ip_stats.get("OutDiscards", 0) - - established = 0 - try: - with open("/proc/net/tcp", "r") as f: - for line in f.readlines()[1:]: - parts = line.strip().split() - if len(parts) >= 4 and parts[3] == "01": - established += 1 - except OSError: - established = 0 - - prev_ts = NETWORK_STACK_PREV["timestamp"] - dt = max(0.001, now - prev_ts) if prev_ts else 1.0 - - def rate(curr, prev): - if prev is None: - return 0.0 - return max(0.0, (curr - prev) / dt) - - retrans_per_sec = rate(retrans_total, NETWORK_STACK_PREV["tcpext_retrans"]) - ip_in_per_sec = rate(ip_in_total, NETWORK_STACK_PREV["ip_in"]) - ip_out_per_sec = rate(ip_out_total, NETWORK_STACK_PREV["ip_out"]) - ip_drop_per_sec = rate(ip_discards_total, NETWORK_STACK_PREV["ip_discards"]) - - rx_per_sec = 0.0 - tx_per_sec = 0.0 - iface_drop_per_sec = 0.0 - rx_bytes = iface_stats.bytes_recv if iface_stats else 0 - tx_bytes = iface_stats.bytes_sent if iface_stats else 0 - iface_drops = (iface_stats.dropin + iface_stats.dropout) if iface_stats else 0 - if NETWORK_STACK_PREV["iface_rx"] is not None: - rx_per_sec = max(0.0, (rx_bytes - NETWORK_STACK_PREV["iface_rx"]) / dt) - if NETWORK_STACK_PREV["iface_tx"] is not None: - tx_per_sec = max(0.0, (tx_bytes - NETWORK_STACK_PREV["iface_tx"]) / dt) - if NETWORK_STACK_PREV["iface_drops"] is not None: - iface_drop_per_sec = max(0.0, (iface_drops - NETWORK_STACK_PREV["iface_drops"]) / dt) - - NETWORK_STACK_PREV["timestamp"] = now - NETWORK_STACK_PREV["tcpext_retrans"] = retrans_total - NETWORK_STACK_PREV["ip_in"] = ip_in_total - NETWORK_STACK_PREV["ip_out"] = ip_out_total - NETWORK_STACK_PREV["ip_discards"] = ip_discards_total - NETWORK_STACK_PREV["iface_rx"] = rx_bytes - NETWORK_STACK_PREV["iface_tx"] = tx_bytes - NETWORK_STACK_PREV["iface_drops"] = iface_drops - - packets_per_sec = ip_in_per_sec + ip_out_per_sec - drop_ratio = (ip_drop_per_sec / packets_per_sec) if packets_per_sec > 0 else 0.0 - throughput_mb_s = (rx_per_sec + tx_per_sec) / (1024 * 1024) - - retrans_prob = min(0.75, retrans_per_sec / 600.0) - drop_prob = min(0.75, (ip_drop_per_sec / 500.0) + (drop_ratio * 8.0)) - packet_speed = max(1.4, min(4.8, 1.8 + throughput_mb_s / 8.0)) - - socket_activity = min(1.0, (len(all_connections) / 80.0) + (retrans_per_sec / 250.0)) - tcp_activity = min(1.0, (ss_metrics.get("cwnd", 0) / 80.0) + (retrans_per_sec / 300.0)) - ip_activity = min(1.0, packets_per_sec / 15000.0) - netfilter_activity = min(1.0, (ip_drop_per_sec / 120.0) + (drop_ratio * 6.0)) - driver_activity = min(1.0, ((rx_per_sec + tx_per_sec) / (60 * 1024 * 1024)) + (iface_drop_per_sec / 40.0)) - nic_activity = min(1.0, ((rx_per_sec + tx_per_sec) / (80 * 1024 * 1024))) - - return { - "timestamp": datetime.now().isoformat(), - "flow": flow, - "layer_metrics": { - "userspace": { - "active_processes": len(psutil.pids()) - }, - "socket_api": { - "active_sockets": len(all_connections), - "established": established, - "retransmits_per_sec": round(retrans_per_sec, 2) - }, - "tcp_udp": { - "established": established, - "retrans_per_sec": round(retrans_per_sec, 2), - "cwnd": int(ss_metrics.get("cwnd", 0)), - "rtt_ms": round(float(ss_metrics.get("rtt_ms", 0.0)), 2), - "tx_queue": int(ss_metrics.get("tx_queue", 0)) - }, - "ip": { - "in_packets_per_sec": round(ip_in_per_sec, 2), - "out_packets_per_sec": round(ip_out_per_sec, 2), - "drop_per_sec": round(ip_drop_per_sec, 3), - "drop_ratio": round(drop_ratio, 5) - }, - "netfilter": { - "drop_per_sec": round(ip_drop_per_sec, 3), - "drop_ratio": round(drop_ratio, 5) - }, - "driver": { - "iface": iface, - "rx_mb_s": round(rx_per_sec / (1024 * 1024), 3), - "tx_mb_s": round(tx_per_sec / (1024 * 1024), 3), - "tx_queue": int(ss_metrics.get("tx_queue", 0)), - "drops_per_sec": round(iface_drop_per_sec, 3) - }, - "nic": { - "iface": iface, - "rx_errors": int(getattr(iface_stats, "errin", 0)) if iface_stats else 0, - "tx_errors": int(getattr(iface_stats, "errout", 0)) if iface_stats else 0, - "drops_total": int(iface_drops) - } - }, - "layer_activity": { - "userspace": min(1.0, len(psutil.pids()) / 400.0), - "socket": round(socket_activity, 4), - "tcp": round(tcp_activity, 4), - "ip": round(ip_activity, 4), - "netfilter": round(netfilter_activity, 4), - "driver": round(driver_activity, 4), - "nic": round(nic_activity, 4) - }, - "signals": { - "drop_probability": round(drop_prob, 4), - "retransmit_probability": round(retrans_prob, 4), - "packet_speed": round(packet_speed, 3) - }, - "throughput_mb_s": round(throughput_mb_s, 3), - "tcp_counters": { - "in_segs": int(tcp_stats.get("InSegs", 0)), - "out_segs": int(tcp_stats.get("OutSegs", 0)), - "retrans_segs_total": int(retrans_total) - } - } + return _network_service.get_network_stack_realtime() def _parse_cgroup_path(pid): - cgroup_text = _proc_fs.safe_read_text(f"/proc/{pid}/cgroup") - if not cgroup_text: - return "/" - chosen = "/" - for line in cgroup_text.splitlines(): - parts = line.split(":") - if len(parts) != 3: - continue - _, controllers, path = parts - path = path.strip() or "/" - # Prefer cgroup v2 unified hierarchy entry "0::/path" - if controllers == "": - return path - if path and path != "/": - chosen = path - return chosen + return _system_view_service._parse_cgroup_path(pid) def _read_namespace_inode(pid, ns_name): - ns_link = f"/proc/{pid}/ns/{ns_name}" - try: - target = os.readlink(ns_link) - except (OSError, PermissionError): - return None - match = re.search(r'\[(\d+)\]', target) - return match.group(1) if match else target + return _system_view_service._read_namespace_inode(pid, ns_name) def _read_cgroup_v2_stats(cgroup_path): - root = "/sys/fs/cgroup" - rel = cgroup_path.lstrip("/") - base = os.path.join(root, rel) if rel else root - - cpu_max_text = _proc_fs.safe_read_text(os.path.join(base, "cpu.max")) - cpu_quota_cores = None - if cpu_max_text: - parts = cpu_max_text.split() - if len(parts) >= 2 and parts[0] != "max": - try: - quota = float(parts[0]) - period = float(parts[1]) - if period > 0: - cpu_quota_cores = round(quota / period, 2) - except ValueError: - cpu_quota_cores = None - - mem_current = _proc_fs.safe_read_text(os.path.join(base, "memory.current")) - mem_max = _proc_fs.safe_read_text(os.path.join(base, "memory.max")) - pids_current = _proc_fs.safe_read_text(os.path.join(base, "pids.current")) - pids_max = _proc_fs.safe_read_text(os.path.join(base, "pids.max")) - io_stat_text = _proc_fs.safe_read_text(os.path.join(base, "io.stat")) - - memory_current_mb = None - memory_max_mb = None - try: - if mem_current is not None: - memory_current_mb = round(int(mem_current) / (1024 * 1024), 1) - except ValueError: - memory_current_mb = None - - try: - if mem_max and mem_max != "max": - memory_max_mb = round(int(mem_max) / (1024 * 1024), 1) - except ValueError: - memory_max_mb = None - - io_bytes = None - if io_stat_text: - total = 0 - for line in io_stat_text.splitlines(): - rbytes_match = re.search(r'rbytes=(\d+)', line) - wbytes_match = re.search(r'wbytes=(\d+)', line) - if rbytes_match: - total += int(rbytes_match.group(1)) - if wbytes_match: - total += int(wbytes_match.group(1)) - io_bytes = total - - return { - "cpu_quota_cores": cpu_quota_cores, - "memory_current_mb": memory_current_mb, - "memory_max_mb": memory_max_mb, - "pids_current": int(pids_current) if pids_current and pids_current.isdigit() else None, - "pids_max": None if pids_max in (None, "max") else (int(pids_max) if pids_max.isdigit() else None), - "io_total_mb": round(io_bytes / (1024 * 1024), 1) if io_bytes is not None else None - } + return _system_view_service._read_cgroup_v2_stats(cgroup_path) def get_isolation_context(): """Aggregate namespace and cgroup context for UI layer.""" - namespace_keys = ["mnt", "pid", "net", "ipc", "uts", "user"] - namespace_labels = { - "mnt": "MNT", - "pid": "PID", - "net": "NET", - "ipc": "IPC", - "uts": "UTS", - "user": "USER" - } - namespace_counts = {k: {} for k in namespace_keys} - cgroup_aggregates = {} - total_scanned = 0 - - for proc in psutil.process_iter(["pid", "name", "memory_info"]): - try: - pid = proc.info["pid"] - total_scanned += 1 - - cgroup_path = _parse_cgroup_path(pid) - agg = cgroup_aggregates.setdefault(cgroup_path, { - "path": cgroup_path, - "process_count": 0, - "memory_mb_sum": 0.0, - "sample_processes": [] - }) - agg["process_count"] += 1 - - mem_info = proc.info.get("memory_info") - if mem_info: - agg["memory_mb_sum"] += (mem_info.rss / (1024 * 1024)) - - if len(agg["sample_processes"]) < 4: - process_name = proc.info.get("name") or "unknown" - agg["sample_processes"].append(process_name) - - for ns_name in namespace_keys: - inode = _read_namespace_inode(pid, ns_name) - if inode: - ns_map = namespace_counts[ns_name] - ns_map[inode] = ns_map.get(inode, 0) + 1 - except (psutil.NoSuchProcess, psutil.AccessDenied, KeyError): - continue - - namespaces = [] - for ns_name in namespace_keys: - entries = namespace_counts[ns_name] - unique_count = len(entries) - dominant_inode = None - dominant_count = 0 - if entries: - dominant_inode, dominant_count = max(entries.items(), key=lambda kv: kv[1]) - activity = round((dominant_count / total_scanned), 3) if total_scanned > 0 else 0 - namespaces.append({ - "id": ns_name, - "label": namespace_labels[ns_name], - "unique_count": unique_count, - "dominant_inode": dominant_inode, - "dominant_count": dominant_count, - "activity": activity - }) - - top_cgroups = sorted( - cgroup_aggregates.values(), - key=lambda x: (x["process_count"], x["memory_mb_sum"]), - reverse=True - )[:4] - - for item in top_cgroups: - stats = _read_cgroup_v2_stats(item["path"]) - item["memory_mb_sum"] = round(item["memory_mb_sum"], 1) - item.update(stats) - - return { - "timestamp": datetime.now().isoformat(), - "processes_scanned": total_scanned, - "namespaces": namespaces, - "top_cgroups": top_cgroups - } + return _system_view_service.get_isolation_context() def get_route_hint(remote_ip): """Fallback path hint using Linux routing table when traceroute tools are absent.""" - ip_cmd = resolve_binary("ip") - if not ip_cmd: - return { - "remote_ip": remote_ip, - "tool": None, - "reached": False, - "hop_count": 0, - "hops": [], - "note": "Path tools unavailable on host" - } - - try: - result = subprocess.run( - [ip_cmd, "-o", "route", "get", remote_ip], - capture_output=True, - text=True, - timeout=2, - check=False - ) - line = (result.stdout or "").strip() - if not line: - return { - "remote_ip": remote_ip, - "tool": "ip-route", - "reached": False, - "hop_count": 0, - "hops": [], - "note": "No route information available" - } - - via_match = re.search(r'\svia\s(\d{1,3}(?:\.\d{1,3}){3})', line) - dev_match = re.search(r'\sdev\s([A-Za-z0-9_.:-]+)', line) - src_match = re.search(r'\ssrc\s(\d{1,3}(?:\.\d{1,3}){3})', line) - - hops = [] - if via_match: - hops.append({ - "hop": 1, - "target": via_match.group(1), - "rtt_ms": None - }) - hops.append({ - "hop": 2, - "target": remote_ip, - "rtt_ms": None - }) - else: - hops.append({ - "hop": 1, - "target": remote_ip, - "rtt_ms": None - }) - - note_parts = ["Traceroute not installed, showing kernel route hint"] - if dev_match: - note_parts.append(f"dev={dev_match.group(1)}") - if src_match: - note_parts.append(f"src={src_match.group(1)}") - - return { - "remote_ip": remote_ip, - "tool": "ip-route", - "reached": False, - "hop_count": len(hops), - "hops": hops, - "note": ", ".join(note_parts) - } - except (subprocess.TimeoutExpired, OSError): - return { - "remote_ip": remote_ip, - "tool": "ip-route", - "reached": False, - "hop_count": 0, - "hops": [], - "note": "Route hint lookup timed out" - } + return _network_service.get_route_hint(remote_ip) def get_traceroute_info(remote_ip, max_hops=8): """Get traceroute/tracepath information for a remote IP with short timeout.""" - try: - target_ip = ipaddress.ip_address(remote_ip) - # Skip loopback/local addresses - traceroute is not meaningful here. - if target_ip.is_loopback or target_ip.is_unspecified: - return { - "remote_ip": remote_ip, - "tool": None, - "reached": False, - "hop_count": 0, - "hops": [], - "note": "Local address, traceroute skipped" - } - except ValueError: - raise ValueError("Invalid IP address") - - now = time.time() - cached = TRACEROUTE_CACHE.get(remote_ip) - if cached and (now - cached["timestamp"]) < TRACEROUTE_CACHE_TTL_SECONDS: - return cached["data"] - - traceroute_cmd = resolve_binary("traceroute") - tracepath_cmd = resolve_binary("tracepath") - cmd = None - tool = None - if traceroute_cmd: - cmd = [traceroute_cmd, "-n", "-m", str(max_hops), "-q", "1", "-w", "1", remote_ip] - tool = "traceroute" - elif tracepath_cmd: - cmd = [tracepath_cmd, "-n", "-m", str(max_hops), remote_ip] - tool = "tracepath" - else: - data = get_route_hint(remote_ip) - TRACEROUTE_CACHE[remote_ip] = {"timestamp": now, "data": data} - return data - - try: - result = subprocess.run( - cmd, - capture_output=True, - text=True, - timeout=7, - check=False - ) - output = (result.stdout or "").strip() - if not output and result.stderr: - output = result.stderr.strip() - except subprocess.TimeoutExpired: - return { - "remote_ip": remote_ip, - "tool": tool, - "reached": False, - "hop_count": 0, - "hops": [], - "note": "Traceroute timed out" - } - - hops = [] - for raw_line in output.splitlines(): - line = raw_line.strip() - hop_match = re.match(r'^(\d+)\s+', line) - if not hop_match: - tracepath_match = re.match(r'^(\d+):\s+', line) - if not tracepath_match: - continue - hop_idx = int(tracepath_match.group(1)) - else: - hop_idx = int(hop_match.group(1)) - - if "*" in line and re.search(r'\*\s*\*\s*\*', line): - hops.append({ - "hop": hop_idx, - "target": "*", - "rtt_ms": None - }) - continue - - ip_match = re.search(r'(\d{1,3}(?:\.\d{1,3}){3})', line) - rtt_match = re.search(r'(\d+(?:\.\d+)?)\s*ms', line) - hops.append({ - "hop": hop_idx, - "target": ip_match.group(1) if ip_match else "?", - "rtt_ms": float(rtt_match.group(1)) if rtt_match else None - }) - - reached = any(h.get("target") == remote_ip for h in hops) - data = { - "remote_ip": remote_ip, - "tool": tool, - "reached": reached, - "hop_count": len(hops), - "hops": hops[:max_hops], - "note": None - } - TRACEROUTE_CACHE[remote_ip] = {"timestamp": now, "data": data} - return data + return _network_service.get_traceroute_info(remote_ip, max_hops=max_hops) def active_connections(): """API for active network connections""" @@ -2160,68 +1282,7 @@ def get_process_fds(pid): def get_processes_detailed(): """API for getting all processes with detailed information (threads, CPU, FDs)""" try: - processes = [] - for proc in psutil.process_iter(['pid', 'name', 'status', 'memory_info', 'cpu_percent', 'num_threads', 'num_fds']): - try: - memory_info = proc.info.get('memory_info') - if memory_info is None: - # Some transient/zombie processes may have incomplete info. - continue - memory_mb = float(memory_info.rss) / 1024 / 1024 - - # Get num_fds with fallback - num_fds = proc.info.get('num_fds') - if num_fds is None or num_fds == 0: - # Try to count from /proc/[pid]/fd directly - try: - pid = proc.info['pid'] - fd_dir = f'/proc/{pid}/fd' - if os.path.exists(fd_dir): - num_fds = len([f for f in os.listdir(fd_dir) if f.isdigit()]) - else: - num_fds = 0 - except (OSError, PermissionError): - num_fds = 0 - if num_fds is None: - num_fds = 0 - - # Get process name - use cmdline for nginx to get full name like "nginx: master process" - process_name = proc.info.get('name') or f'pid-{proc.info.get("pid", "unknown")}' - try: - cmdline = proc.cmdline() - if cmdline and len(cmdline) > 0: - # For nginx, cmdline[0] is "nginx:" and we want the full description - if cmdline[0] == 'nginx:' and len(cmdline) > 1: - process_name = f"nginx: {cmdline[1]}" - except (psutil.AccessDenied, psutil.NoSuchProcess): - pass - - # Get cmdline for better process identification - cmdline_str = '' - try: - cmdline = proc.cmdline() - if cmdline: - cmdline_str = ' '.join(cmdline) - except (psutil.AccessDenied, psutil.NoSuchProcess): - pass - - processes.append({ - 'pid': proc.info['pid'], - 'name': process_name, - 'cmdline': cmdline_str, # Add cmdline for better identification - 'status': proc.info.get('status', 'unknown'), - 'memory_mb': round(memory_mb, 1), - 'cpu_percent': round(float(proc.info.get('cpu_percent', 0) or 0), 1), - 'num_threads': int(proc.info.get('num_threads', 0) or 0), - 'num_fds': int(num_fds or 0) - }) - except (psutil.NoSuchProcess, psutil.AccessDenied, psutil.ZombieProcess): - continue - except Exception: - # Never fail the whole endpoint because of one malformed/transient process. - continue - - return jsonify({'processes': processes}) + return jsonify({"processes": _processes_service.get_processes_detailed_data()}) except Exception as e: return jsonify({'error': str(e)}), 500 @@ -4374,501 +3435,19 @@ def _read_text(path): } def _parse_meminfo_kb(): - """Linux /proc/meminfo values in kB (same units as psutil docs).""" - out = {} - try: - with open("/proc/meminfo", "r", encoding="utf-8", errors="ignore") as f: - for line in f: - parts = line.split() - if len(parts) >= 2 and parts[1].isdigit(): - out[parts[0].rstrip(":")] = int(parts[1]) - except Exception: - pass - return out + return _processes_service._parse_meminfo_kb() def _memory_strip_blocks(kind, kb_k, mem_total_kb, n_blocks, seed0): - """Variable-width blocks in one horizontal strip; heat ~ share of RAM in this category.""" - mem_total_kb = max(1, int(mem_total_kb)) - kb_k = max(0, int(kb_k)) - weights = [] - for i in range(n_blocks): - v = (((seed0 + i * 104729) % 1000) + 40) / 1040.0 - weights.append(v) - sw = sum(weights) - share = kb_k / float(mem_total_kb) - blocks = [] - for i in range(n_blocks): - w = weights[i] / sw - heat = min( - 1.0, - 0.05 + min(0.92, share * 2.0) + (((seed0 + i * 31) % 15) / 120.0), - ) - blocks.append({"w": round(w, 6), "heat": round(heat, 4), "kind": kind}) - return blocks + return _processes_service._memory_strip_blocks(kind, kb_k, mem_total_kb, n_blocks, seed0) def _build_memory_visual_rows(meminfo_kb, syscall_nodes, vm, swap): - """ - Rows of horizontal strips: each row ≈ one kernel/accounting bucket from /proc/meminfo. - Block widths are stylistic subdivisions; row mass is proportional to kb / MemTotal. - """ - mi = meminfo_kb or {} - mt = max(1, int(mi.get("MemTotal") or 0)) - if mt <= 1: - try: - mt = max(1, int(getattr(vm, "total", 0) / 1024)) - except Exception: - mt = 1 - - seed_base = (mt % 100000) + int(mi.get("Active", 0) or 0) % 50000 - - row_specs = [ - ("buffers", "buffers", mi.get("Buffers", 0)), - ("cached", "page cache", mi.get("Cached", 0)), - ("anon", "anonymous (heap/stack)", mi.get("AnonPages", 0)), - ] - # Slab: split reclaimable vs unreclaimable when both exist (Linux 2.6.19+). - if mi.get("SReclaimable") is not None and mi.get("SUnreclaim") is not None: - row_specs.append(("sreclaim", "slab reclaimable", mi.get("SReclaimable", 0))) - row_specs.append(("sunreclaim", "slab unreclaimable", mi.get("SUnreclaim", 0))) - else: - row_specs.append(("slab", "slab / kmalloc", mi.get("Slab", 0))) - row_specs.extend( - [ - ("shmem", "shmem / tmpfs", mi.get("Shmem", 0)), - ("mapped", "file mappings", mi.get("Mapped", 0)), - ] - ) - dirty_wb = int(mi.get("Dirty", 0) or 0) + int(mi.get("Writeback", 0) or 0) + int( - mi.get("WritebackTmp", 0) or 0 - ) - if dirty_wb > 0: - row_specs.append(("dirty_wb", "dirty + writeback", dirty_wb)) - ah = int(mi.get("AnonHugePages", 0) or 0) - if ah > 0: - row_specs.append(("anon_huge", "transparent huge pages (anon)", ah)) - shm_h = int(mi.get("ShmemHugePages", 0) or 0) - if shm_h > 0: - row_specs.append(("shmem_huge", "huge pages (shmem)", shm_h)) - vmu = int(mi.get("VmallocUsed", 0) or 0) - if vmu > 0: - row_specs.append(("vmalloc", "vmalloc used", vmu)) - ac = int(mi.get("Active", 0) or 0) - iac = int(mi.get("Inactive", 0) or 0) - if ac > 0: - row_specs.append(("active", "active (LRU)", ac)) - if iac > 0: - row_specs.append(("inactive", "inactive (LRU)", iac)) - swap_tot = int(mi.get("SwapTotal", 0) or 0) - swap_free = int(mi.get("SwapFree", 0) or 0) - swap_used = max(0, swap_tot - swap_free) - if swap_tot > 0: - row_specs.append(("swap", "swap occupied", swap_used)) - - pt = int(mi.get("PageTables", 0) or 0) - ks = int(mi.get("KernelStack", 0) or 0) - if pt + ks > 0: - row_specs.append(("kmeta", "pagetables + kernel stacks", pt + ks)) - - rows = [] - for sk, label, kb in row_specs: - kb = int(kb or 0) - if kb <= 0 and sk != "swap": - continue - if sk == "swap" and kb <= 0: - continue - sk_seed = sum(ord(c) for c in sk) * 31 + len(sk) - n_blocks = 22 + (seed_base % 11) + (sk_seed % 9) - seed0 = seed_base + (sk_seed % 100000) - blocks = _memory_strip_blocks(sk, kb, mt, n_blocks, seed0) - rows.append( - { - "id": sk, - "label": label, - "kb": kb, - "pct_of_ram": round(100.0 * kb / float(mt), 2) if mt else 0.0, - "blocks": blocks, - } - ) - - top_tasks = sorted( - syscall_nodes, - key=lambda x: int(x.get("rss_bytes") or 0), - reverse=True, - )[:6] - if top_tasks: - tr_bytes = sum(int(x.get("rss_bytes") or 0) for x in top_tasks) or 1 - tr_kb = max(1, int(tr_bytes / 1024)) - task_blocks = [] - for p in top_tasks: - rss = int(p.get("rss_bytes") or 0) - if rss <= 0: - continue - w = rss / float(tr_bytes) - mp = float(p.get("memory_percent") or 0.0) - heat = min(1.0, 0.2 + (mp / 100.0) * 0.75 + (rss / float(tr_bytes)) * 0.15) - task_blocks.append( - { - "w": round(w, 6), - "heat": round(heat, 4), - "kind": "task", - "pid": int(p.get("pid") or 0), - "name": str(p.get("name") or "")[:14], - } - ) - if task_blocks: - sw = sum(b["w"] for b in task_blocks) - if sw > 0: - for b in task_blocks: - b["w"] = round(b["w"] / sw, 6) - rows.append( - { - "id": "tasks", - "label": "sampled tasks RSS (top)", - "kb": tr_kb, - "pct_of_ram": round(100.0 * tr_kb / float(mt), 2) if mt else 0.0, - "blocks": task_blocks, - } - ) - - dirty_kb = int(mi.get("Dirty", 0) or 0) - wb_kb = int(mi.get("Writeback", 0) or 0) - sr_kb = int(mi.get("SReclaimable", 0) or 0) - su_kb = int(mi.get("SUnreclaim", 0) or 0) - slab_total_kb = int(mi.get("Slab", 0) or 0) or (sr_kb + su_kb) - summary = { - "total_mb": round(mt / 1024.0, 1), - "used_percent": round(vm.percent, 1) if vm else 0.0, - "available_mb": round((mi.get("MemAvailable", 0) or 0) / 1024.0, 1), - "swap_percent": round(swap.percent, 1) if swap else 0.0, - "buffers_mb": round((mi.get("Buffers", 0) or 0) / 1024.0, 1), - "cached_mb": round((mi.get("Cached", 0) or 0) / 1024.0, 1), - "anon_mb": round((mi.get("AnonPages", 0) or 0) / 1024.0, 1), - "slab_mb": round(slab_total_kb / 1024.0, 1), - "sreclaimable_mb": round(sr_kb / 1024.0, 1), - "sunreclaim_mb": round(su_kb / 1024.0, 1), - "dirty_mb": round(dirty_kb / 1024.0, 2), - "writeback_mb": round(wb_kb / 1024.0, 2), - "dirty_writeback_mb": round(dirty_wb / 1024.0, 2), - "anon_huge_mb": round(ah / 1024.0, 2), - "shmem_huge_mb": round(shm_h / 1024.0, 2), - "vmalloc_mb": round(vmu / 1024.0, 2), - "active_mb": round(ac / 1024.0, 1), - "inactive_mb": round(iac / 1024.0, 1), - "swap_used_mb": round(swap_used / 1024.0, 1) if swap_tot else 0.0, - "source": "proc_meminfo+psutil+v2", - } - return rows, summary + return _processes_service._build_memory_visual_rows(meminfo_kb, syscall_nodes, vm, swap) def collect_processes_realtime(): - """ - Processes subsystem telemetry focused on: - - syscall interception signals - - network tracing - - security hooks - """ - lsm_raw = "" - try: - with open("/sys/kernel/security/lsm", "r", encoding="utf-8", errors="ignore") as f: - lsm_raw = str(f.read().strip()) - except Exception: - lsm_raw = "" - active_lsms = [x.strip() for x in lsm_raw.split(",") if x.strip()] - - yama_scope = "" - try: - with open("/proc/sys/kernel/yama/ptrace_scope", "r", encoding="utf-8", errors="ignore") as f: - yama_scope = str(f.read().strip()) - except Exception: - yama_scope = "" - - syscall_nodes = [] - seccomp_modes = {"none": 0, "strict": 0, "filter": 0, "unknown": 0} - for proc in psutil.process_iter(["pid", "ppid", "name", "username", "cpu_percent", "memory_percent", "num_threads"]): - try: - pid = int(proc.info.get("pid") or 0) - if pid <= 0: - continue - ppid = int(proc.info.get("ppid") or 0) - name = str(proc.info.get("name") or "unknown") - user = str(proc.info.get("username") or "") - cpu = float(proc.info.get("cpu_percent") or 0.0) - mem = float(proc.info.get("memory_percent") or 0.0) - threads = int(proc.info.get("num_threads") or 0) - rss = 0 - try: - rss = int(getattr(proc.memory_info(), "rss", 0) or 0) - except Exception: - rss = 0 - fd_count = 0 - try: - fd_count = int(proc.num_fds() or 0) - except Exception: - fd_count = 0 - seccomp_mode = "unknown" - with open(f"/proc/{pid}/status", "r", encoding="utf-8", errors="ignore") as f: - for ln in f: - if ln.startswith("Seccomp:"): - raw = ln.split(":", 1)[1].strip() - if raw == "0": - seccomp_mode = "none" - elif raw == "1": - seccomp_mode = "strict" - elif raw == "2": - seccomp_mode = "filter" - else: - seccomp_mode = "unknown" - break - seccomp_modes[seccomp_mode] = seccomp_modes.get(seccomp_mode, 0) + 1 - syscall_pressure = min(100, int(cpu * 1.5 + threads * 0.35 + mem * 0.8)) - syscall_nodes.append({ - "pid": pid, - "ppid": ppid, - "name": name, - "user": user, - "fd_count": fd_count, - "syscall_pressure": syscall_pressure, - "seccomp_mode": seccomp_mode, - "memory_percent": round(mem, 2), - "rss_bytes": rss, - }) - except Exception: - continue - syscall_nodes.sort(key=lambda x: x.get("syscall_pressure", 0), reverse=True) - syscall_nodes = syscall_nodes[:14] - - network_nodes = {} - try: - for conn in psutil.net_connections(kind="inet"): - pid = int(getattr(conn, "pid", 0) or 0) - if pid <= 0: - continue - remote_ip = "" - try: - raddr = getattr(conn, "raddr", None) - if raddr and len(raddr) >= 1: - remote_ip = str(raddr[0]) - except Exception: - remote_ip = "" - status = str(getattr(conn, "status", "") or "").upper() - bucket = network_nodes.get(pid) - if not bucket: - proc_name = "unknown" - try: - proc_name = psutil.Process(pid).name() - except Exception: - proc_name = "unknown" - bucket = { - "pid": pid, - "name": proc_name, - "connections": 0, - "remote_ips": set(), - "states": {} - } - network_nodes[pid] = bucket - bucket["connections"] += 1 - if remote_ip: - bucket["remote_ips"].add(remote_ip) - if status: - bucket["states"][status] = bucket["states"].get(status, 0) + 1 - except Exception: - pass - - network_tracing = [] - for _, row in network_nodes.items(): - states_sorted = sorted(row["states"].items(), key=lambda kv: kv[1], reverse=True) - top_state = states_sorted[0][0] if states_sorted else "UNKNOWN" - network_tracing.append({ - "pid": int(row["pid"]), - "name": str(row["name"]), - "connections": int(row["connections"]), - "unique_peers": int(len(row["remote_ips"])), - "peer_sample": sorted(list(row["remote_ips"]))[:4], - "top_state": top_state - }) - network_tracing.sort(key=lambda x: (x.get("connections", 0), x.get("unique_peers", 0)), reverse=True) - network_tracing = network_tracing[:14] - - security_hooks = [ - { - "name": "LSM stack", - "status": "active" if active_lsms else "unknown", - "detail": ",".join(active_lsms[:4]) if active_lsms else "n/a" - }, - { - "name": "SELinux/AppArmor engines", - "status": "active" if any(x in {"selinux", "apparmor"} for x in active_lsms) else "inactive", - "detail": "policy-enforcement-path" - }, - { - "name": "BPF LSM", - "status": "active" if "bpf" in active_lsms else "inactive", - "detail": "dynamic-policy-hook" - }, - { - "name": "seccomp filter gate", - "status": "active" if (seccomp_modes.get("filter", 0) + seccomp_modes.get("strict", 0)) > 0 else "inactive", - "detail": f"filter:{seccomp_modes.get('filter', 0)} strict:{seccomp_modes.get('strict', 0)}" - }, - { - "name": "Yama ptrace scope", - "status": "hardened" if yama_scope in {"2", "3"} else ("relaxed" if yama_scope in {"0", "1"} else "unknown"), - "detail": yama_scope or "n/a" - } - ] - - # Neural graph model: nodes=processes, edges=behavior interactions. - node_pool = {} - for row in syscall_nodes[:16]: - pid = int(row.get("pid") or 0) - if pid <= 0: - continue - node_pool[pid] = { - "pid": pid, - "ppid": int(row.get("ppid") or 0), - "name": str(row.get("name") or "unknown"), - "user": str(row.get("user") or ""), - "syscall_pressure": int(row.get("syscall_pressure") or 0), - "fd_count": int(row.get("fd_count") or 0), - "seccomp_mode": str(row.get("seccomp_mode") or "unknown"), - "connections": 0, - "unique_peers": 0, - "memory_percent": float(row.get("memory_percent") or 0.0), - "rss_bytes": int(row.get("rss_bytes") or 0), - } - for row in network_tracing[:16]: - pid = int(row.get("pid") or 0) - if pid <= 0: - continue - if pid not in node_pool: - node_pool[pid] = { - "pid": pid, - "ppid": 0, - "name": str(row.get("name") or "unknown"), - "user": "", - "syscall_pressure": 0, - "fd_count": 0, - "seccomp_mode": "unknown", - "connections": 0, - "unique_peers": 0, - "memory_percent": 0.0, - "rss_bytes": 0, - } - node_pool[pid]["connections"] = int(row.get("connections") or 0) - node_pool[pid]["unique_peers"] = int(row.get("unique_peers") or 0) - - edges = [] - edge_keys = set() - network_by_pid = {int(r.get("pid") or 0): r for r in network_tracing} - node_pids = sorted(node_pool.keys()) - - def _add_edge(src_pid, dst_pid, edge_type, weight): - src = int(src_pid or 0) - dst = int(dst_pid or 0) - if src <= 0 or dst <= 0 or src == dst: - return - if src not in node_pool or dst not in node_pool: - return - pair = tuple(sorted((src, dst))) - key = (pair[0], pair[1], edge_type) - if key in edge_keys: - return - edge_keys.add(key) - edges.append({ - "source": src, - "target": dst, - "type": edge_type, - "weight": float(max(0.1, min(1.0, weight))) - }) - - # IPC edges: parent-child links inside the sampled set. - for pid, node in node_pool.items(): - ppid = int(node.get("ppid") or 0) - if ppid in node_pool: - _add_edge(pid, ppid, "ipc", 0.72) - - # Syscalls edges: close-pressure processes likely competing on kernel hooks. - sorted_by_pressure = sorted(node_pool.values(), key=lambda n: n.get("syscall_pressure", 0), reverse=True) - for i in range(len(sorted_by_pressure) - 1): - a = sorted_by_pressure[i] - b = sorted_by_pressure[i + 1] - diff = abs(int(a.get("syscall_pressure", 0)) - int(b.get("syscall_pressure", 0))) - weight = 1.0 - min(0.8, diff / 100.0) - _add_edge(int(a.get("pid")), int(b.get("pid")), "syscalls", weight) - - # Network edges: connect nodes that share at least one peer sample. - for i in range(len(node_pids)): - for j in range(i + 1, len(node_pids)): - pa = node_pids[i] - pb = node_pids[j] - ra = network_by_pid.get(pa) or {} - rb = network_by_pid.get(pb) or {} - sa = set(ra.get("peer_sample") or []) - sb = set(rb.get("peer_sample") or []) - if sa and sb and (sa & sb): - _add_edge(pa, pb, "network", 0.88) - - # File access edges: processes with high FD count and same user. - for i in range(len(node_pids)): - for j in range(i + 1, len(node_pids)): - na = node_pool[node_pids[i]] - nb = node_pool[node_pids[j]] - if not na.get("user") or na.get("user") != nb.get("user"): - continue - fa = int(na.get("fd_count") or 0) - fb = int(nb.get("fd_count") or 0) - if fa >= 16 and fb >= 16: - _add_edge(int(na.get("pid")), int(nb.get("pid")), "file_access", 0.64) - - nodes = list(node_pool.values())[:18] - edges = edges[:64] - - try: - vm = psutil.virtual_memory() - swap = psutil.swap_memory() - meminfo_kb = _parse_meminfo_kb() - strip_rows, mem_summary = _build_memory_visual_rows(meminfo_kb, syscall_nodes, vm, swap) - memory_visual = { - "layout": "strips", - "rows": strip_rows, - "summary": mem_summary, - } - except Exception: - memory_visual = { - "layout": "strips", - "rows": [], - "summary": { - "total_mb": 0, - "used_percent": 0.0, - "available_mb": 0, - "swap_percent": 0.0, - "source": "error", - }, - } - - return { - "timestamp": datetime.utcnow().isoformat() + "Z", - "syscalls_interception": syscall_nodes, - "network_tracing": network_tracing, - "security_hooks": security_hooks, - "neural_graph": { - "nodes": nodes, - "edges": edges - }, - "memory_visual": memory_visual, - "meta": { - "processes_sampled": len(syscall_nodes), - "network_processes": len(network_tracing), - "seccomp_filter_percent": round( - (seccomp_modes.get("filter", 0) + seccomp_modes.get("strict", 0)) - * 100.0 / max(1, sum(seccomp_modes.values())), - 2 - ), - "mode": "live-heuristic-v1" - } - } + return _processes_service.collect_processes_realtime() def kernel_dna(): """API endpoint for Kernel DNA visualization data""" @@ -4924,6 +3503,22 @@ def ingest_frontend_logs(): return jsonify({"status": "ok", "accepted": accepted}) +def get_proc_graph(): + """3D /proc graph for ``proc-3d-visualization.js``: nodes with x,y,z and edges {from,to}.""" + try: + return jsonify(_processes_service.get_proc_graph_data()) + except Exception as e: + return jsonify({"error": str(e), "nodes": [], "edges": []}), 500 + + +def get_process_files(): + """Bezier process↔file curves for ``bezier_curves.js`` (optional real data; empty → decorative fallback).""" + try: + return jsonify({"curves": [], "timestamp": datetime.now().isoformat()}) + except Exception as e: + return jsonify({"error": str(e), "curves": []}), 500 + + register_http_routes(app)