"""Process scanner for AYN Antivirus. Inspects running processes for known crypto-miners, anomalous CPU usage, and hidden / stealth processes. Uses ``psutil`` for cross-platform process enumeration and ``/proc`` on Linux for hidden-process detection. """ from __future__ import annotations import logging import os import signal from datetime import datetime from pathlib import Path from typing import Any, Dict, List, Optional import psutil from ayn_antivirus.constants import ( CRYPTO_MINER_PROCESS_NAMES, HIGH_CPU_THRESHOLD, ) from ayn_antivirus.scanners.base import BaseScanner logger = logging.getLogger(__name__) class ProcessScanner(BaseScanner): """Scan running processes for malware, miners, and anomalies. Parameters ---------- cpu_threshold: CPU-usage percentage above which a process is flagged. Defaults to :pydata:`constants.HIGH_CPU_THRESHOLD`. """ def __init__(self, cpu_threshold: float = HIGH_CPU_THRESHOLD) -> None: self.cpu_threshold = cpu_threshold # ------------------------------------------------------------------ # BaseScanner interface # ------------------------------------------------------------------ @property def name(self) -> str: return "process_scanner" @property def description(self) -> str: return "Inspects running processes for miners and suspicious activity" def scan(self, target: Any = None) -> Dict[str, Any]: """Run a full process scan. *target* is ignored — all live processes are inspected. Returns ------- dict ``total``, ``suspicious``, ``high_cpu``, ``hidden``. """ all_procs = self.get_all_processes() suspicious = self.find_suspicious_processes() high_cpu = self.find_high_cpu_processes() hidden = self.find_hidden_processes() return { "total": len(all_procs), "suspicious": suspicious, "high_cpu": high_cpu, "hidden": hidden, } # ------------------------------------------------------------------ # Process enumeration # ------------------------------------------------------------------ @staticmethod def get_all_processes() -> List[Dict[str, Any]]: """Return a snapshot of every running process. Each dict contains: ``pid``, ``name``, ``cmdline``, ``cpu_percent``, ``memory_percent``, ``username``, ``create_time``, ``connections``, ``open_files``. """ result: List[Dict[str, Any]] = [] attrs = [ "pid", "name", "cmdline", "cpu_percent", "memory_percent", "username", "create_time", ] for proc in psutil.process_iter(attrs): try: info = proc.info # Connections and open files are expensive; fetch lazily. try: connections = [ { "fd": c.fd, "family": str(c.family), "type": str(c.type), "laddr": f"{c.laddr.ip}:{c.laddr.port}" if c.laddr else "", "raddr": f"{c.raddr.ip}:{c.raddr.port}" if c.raddr else "", "status": c.status, } for c in proc.net_connections() ] except (psutil.AccessDenied, psutil.NoSuchProcess, OSError): connections = [] try: open_files = [f.path for f in proc.open_files()] except (psutil.AccessDenied, psutil.NoSuchProcess, OSError): open_files = [] create_time = info.get("create_time") result.append({ "pid": info["pid"], "name": info.get("name", ""), "cmdline": info.get("cmdline") or [], "cpu_percent": info.get("cpu_percent") or 0.0, "memory_percent": info.get("memory_percent") or 0.0, "username": info.get("username", "?"), "create_time": ( datetime.utcfromtimestamp(create_time).isoformat() if create_time else None ), "connections": connections, "open_files": open_files, }) except (psutil.NoSuchProcess, psutil.AccessDenied, psutil.ZombieProcess): continue return result # ------------------------------------------------------------------ # Suspicious-process detection # ------------------------------------------------------------------ def find_suspicious_processes(self) -> List[Dict[str, Any]]: """Return processes whose name or command line matches a known miner. Matches are case-insensitive against :pydata:`constants.CRYPTO_MINER_PROCESS_NAMES`. """ suspicious: List[Dict[str, Any]] = [] for proc in psutil.process_iter(["pid", "name", "cmdline", "cpu_percent", "username"]): try: info = proc.info pname = (info.get("name") or "").lower() cmdline = " ".join(info.get("cmdline") or []).lower() for miner in CRYPTO_MINER_PROCESS_NAMES: if miner in pname or miner in cmdline: suspicious.append({ "pid": info["pid"], "name": info.get("name", ""), "cmdline": info.get("cmdline") or [], "cpu_percent": info.get("cpu_percent") or 0.0, "username": info.get("username", "?"), "matched_signature": miner, "reason": f"Known miner process: {miner}", "severity": "CRITICAL", }) break # one match per process except (psutil.NoSuchProcess, psutil.AccessDenied, psutil.ZombieProcess): continue return suspicious # ------------------------------------------------------------------ # High-CPU detection # ------------------------------------------------------------------ def find_high_cpu_processes( self, threshold: Optional[float] = None, ) -> List[Dict[str, Any]]: """Return processes whose CPU usage exceeds *threshold* percent. Parameters ---------- threshold: Override the instance-level ``cpu_threshold``. """ limit = threshold if threshold is not None else self.cpu_threshold high: List[Dict[str, Any]] = [] for proc in psutil.process_iter(["pid", "name", "cmdline", "cpu_percent", "username"]): try: info = proc.info cpu = info.get("cpu_percent") or 0.0 if cpu > limit: high.append({ "pid": info["pid"], "name": info.get("name", ""), "cmdline": info.get("cmdline") or [], "cpu_percent": cpu, "username": info.get("username", "?"), "reason": f"High CPU usage: {cpu:.1f}%", "severity": "HIGH", }) except (psutil.NoSuchProcess, psutil.AccessDenied, psutil.ZombieProcess): continue return high # ------------------------------------------------------------------ # Hidden-process detection (Linux only) # ------------------------------------------------------------------ @staticmethod def find_hidden_processes() -> List[Dict[str, Any]]: """Detect processes visible in ``/proc`` but hidden from ``psutil``. On non-Linux systems this returns an empty list. A mismatch may indicate a userland rootkit that hooks the process listing syscalls. """ proc_dir = Path("/proc") if not proc_dir.is_dir(): return [] # not Linux # PIDs visible via /proc filesystem. proc_pids: set[int] = set() try: for entry in proc_dir.iterdir(): if entry.name.isdigit(): proc_pids.add(int(entry.name)) except PermissionError: logger.warning("Cannot enumerate /proc") return [] # PIDs visible via psutil (which ultimately calls getdents / readdir). psutil_pids = set(psutil.pids()) hidden: List[Dict[str, Any]] = [] for pid in proc_pids - psutil_pids: # Read whatever we can from /proc/. name = "" cmdline = "" try: comm = proc_dir / str(pid) / "comm" if comm.exists(): name = comm.read_text().strip() except OSError: pass try: cl = proc_dir / str(pid) / "cmdline" if cl.exists(): cmdline = cl.read_bytes().replace(b"\x00", b" ").decode(errors="replace").strip() except OSError: pass hidden.append({ "pid": pid, "name": name, "cmdline": cmdline, "reason": "Process visible in /proc but hidden from psutil (possible rootkit)", "severity": "CRITICAL", }) return hidden # ------------------------------------------------------------------ # Single-process detail # ------------------------------------------------------------------ @staticmethod def get_process_details(pid: int) -> Dict[str, Any]: """Return comprehensive information about a single process. Raises ------ psutil.NoSuchProcess If the PID does not exist. psutil.AccessDenied If the caller lacks permission to inspect the process. """ proc = psutil.Process(pid) with proc.oneshot(): info: Dict[str, Any] = { "pid": proc.pid, "name": proc.name(), "exe": "", "cmdline": proc.cmdline(), "status": proc.status(), "username": "", "cpu_percent": proc.cpu_percent(interval=0.1), "memory_percent": proc.memory_percent(), "memory_info": {}, "create_time": datetime.utcfromtimestamp(proc.create_time()).isoformat(), "cwd": "", "open_files": [], "connections": [], "threads": proc.num_threads(), "nice": None, "environ": {}, } try: info["exe"] = proc.exe() except (psutil.AccessDenied, OSError): pass try: info["username"] = proc.username() except psutil.AccessDenied: pass try: mem = proc.memory_info() info["memory_info"] = {"rss": mem.rss, "vms": mem.vms} except (psutil.AccessDenied, OSError): pass try: info["cwd"] = proc.cwd() except (psutil.AccessDenied, OSError): pass try: info["open_files"] = [f.path for f in proc.open_files()] except (psutil.AccessDenied, OSError): pass try: info["connections"] = [ { "laddr": f"{c.laddr.ip}:{c.laddr.port}" if c.laddr else "", "raddr": f"{c.raddr.ip}:{c.raddr.port}" if c.raddr else "", "status": c.status, } for c in proc.net_connections() ] except (psutil.AccessDenied, OSError): pass try: info["nice"] = proc.nice() except (psutil.AccessDenied, OSError): pass try: info["environ"] = dict(proc.environ()) except (psutil.AccessDenied, OSError): pass return info # ------------------------------------------------------------------ # Process control # ------------------------------------------------------------------ @staticmethod def kill_process(pid: int) -> bool: """Send ``SIGKILL`` to the process with *pid*. Returns ``True`` if the signal was delivered successfully, ``False`` otherwise (e.g. the process no longer exists or permission denied). """ try: proc = psutil.Process(pid) proc.kill() # SIGKILL proc.wait(timeout=5) logger.info("Killed process %d (%s)", pid, proc.name()) return True except psutil.NoSuchProcess: logger.warning("Process %d no longer exists", pid) return False except psutil.AccessDenied: logger.error("Permission denied killing process %d", pid) # Fall back to raw signal as a last resort. try: os.kill(pid, signal.SIGKILL) logger.info("Killed process %d via os.kill", pid) return True except OSError as exc: logger.error("os.kill(%d) failed: %s", pid, exc) return False except psutil.TimeoutExpired: logger.warning("Process %d did not exit within timeout", pid) return False