from __future__ import annotations from dataclasses import dataclass import os from pathlib import Path import pwd import time from typing import Mapping from .models import ProcessSample from .sanitize import render_argv, sanitize_argv CLK_TCK = os.sysconf(os.sysconf_names.get("SC_CLK_TCK", "SC_CLK_TCK")) PAGE_SIZE = os.sysconf(os.sysconf_names.get("SC_PAGE_SIZE", "SC_PAGE_SIZE")) @dataclass(frozen=True) class StatFields: pid: int comm: str state: str ppid: int pgid: int sid: int tty_nr: int utime_ticks: int stime_ticks: int priority: int nice: int num_threads: int start_time_ticks: int vms_bytes: int rss_pages: int def parse_stat(stat_text: str) -> StatFields: open_idx = stat_text.find("(") close_idx = stat_text.rfind(")") if open_idx < 0 or close_idx < open_idx: raise ValueError("invalid stat comm field") pid = int(stat_text[:open_idx].strip()) comm = stat_text[open_idx + 1 : close_idx] rest = stat_text[close_idx + 2 :].split() if len(rest) < 22: raise ValueError("stat line is missing fields") return StatFields( pid=pid, comm=comm, state=rest[0], ppid=int(rest[1]), pgid=int(rest[2]), sid=int(rest[3]), tty_nr=int(rest[4]), utime_ticks=int(rest[11]), stime_ticks=int(rest[12]), priority=int(rest[15]), nice=int(rest[16]), num_threads=int(rest[17]), start_time_ticks=int(rest[19]), vms_bytes=int(rest[20]), rss_pages=int(rest[21]), ) def read_boot_id(proc_root: Path = Path("/proc")) -> str: try: return (proc_root / "sys/kernel/random/boot_id").read_text().strip() except OSError: return "unknown-boot" def read_uptime(proc_root: Path = Path("/proc")) -> float: try: return float((proc_root / "uptime").read_text().split()[0]) except (OSError, ValueError, IndexError): return time.monotonic() def _read_status(path: Path) -> tuple[int, str, int, tuple[str, ...]]: uid = -1 user = "" threads = 0 errors: list[str] = [] try: for line in path.read_text(errors="replace").splitlines(): if line.startswith("Uid:"): parts = line.split() uid = int(parts[1]) try: user = pwd.getpwuid(uid).pw_name except KeyError: user = str(uid) elif line.startswith("Threads:"): threads = int(line.split()[1]) except (OSError, ValueError, IndexError) as exc: errors.append(f"status:{type(exc).__name__}") return uid, user, threads, tuple(errors) def _read_cmdline(path: Path, comm: str) -> tuple[tuple[str, ...], str, int, tuple[str, ...]]: errors: list[str] = [] try: raw = path.read_bytes() except OSError as exc: raw = b"" errors.append(f"cmdline:{type(exc).__name__}") argv_raw = [part.decode("utf-8", "replace") for part in raw.split(b"\0") if part] if not argv_raw and comm: argv_raw = [comm] argv, masked_count = sanitize_argv(argv_raw) return argv, render_argv(argv), masked_count, tuple(errors) def _read_statm(path: Path) -> tuple[int, int, tuple[str, ...]]: try: fields = path.read_text().split() vms_pages = int(fields[0]) rss_pages = int(fields[1]) return rss_pages * PAGE_SIZE, vms_pages * PAGE_SIZE, () except (OSError, ValueError, IndexError) as exc: return 0, 0, (f"statm:{type(exc).__name__}",) def _read_text(path: Path, label: str) -> tuple[str, tuple[str, ...]]: try: return path.read_text(errors="replace").strip(), () except OSError as exc: return "", (f"{label}:{type(exc).__name__}",) def _read_link(path: Path, label: str) -> tuple[str, tuple[str, ...]]: try: return os.readlink(path), () except OSError as exc: return "", (f"{label}:{type(exc).__name__}",) def _sample_pid( pid_dir: Path, *, boot_id: str, uptime_seconds: float, sample_ts: float, previous: Mapping[str, ProcessSample], current_uid_only: bool, exclude_pids: set[int], ) -> ProcessSample | None: try: pid = int(pid_dir.name) except ValueError: return None if pid in exclude_pids: return None errors: list[str] = [] try: stat_fields = parse_stat((pid_dir / "stat").read_text()) except (OSError, ValueError): return None uid, user, threads, status_errors = _read_status(pid_dir / "status") errors.extend(status_errors) if current_uid_only and uid != -1 and uid != os.getuid(): return None process_key = f"{boot_id}:{stat_fields.pid}:{stat_fields.start_time_ticks}:{uid}" argv, argv_display, masked_count, cmd_errors = _read_cmdline( pid_dir / "cmdline", stat_fields.comm ) errors.extend(cmd_errors) rss_bytes, vms_bytes, statm_errors = _read_statm(pid_dir / "statm") errors.extend(statm_errors) if not rss_bytes: rss_bytes = max(0, stat_fields.rss_pages) * PAGE_SIZE if not vms_bytes: vms_bytes = stat_fields.vms_bytes cgroup, cgroup_errors = _read_text(pid_dir / "cgroup", "cgroup") errors.extend(cgroup_errors) exe_path, exe_errors = _read_link(pid_dir / "exe", "exe") cwd, cwd_errors = _read_link(pid_dir / "cwd", "cwd") errors.extend(exe_errors) errors.extend(cwd_errors) age = max(0.0, uptime_seconds - (stat_fields.start_time_ticks / CLK_TCK)) cpu_percent = 0.0 previous_sample = previous.get(process_key) if previous_sample is not None: wall_delta = sample_ts - previous_sample.sample_ts ticks_delta = ( stat_fields.utime_ticks + stat_fields.stime_ticks - previous_sample.cpu_ticks ) if wall_delta > 0 and ticks_delta >= 0: cpu_percent = 100.0 * (ticks_delta / CLK_TCK) / wall_delta return ProcessSample( sample_ts=sample_ts, boot_id=boot_id, pid=stat_fields.pid, ppid=stat_fields.ppid, pgid=stat_fields.pgid, sid=stat_fields.sid, uid=uid, user=user or str(uid), comm=stat_fields.comm, state=stat_fields.state, start_time_ticks=stat_fields.start_time_ticks, process_key=process_key, age_seconds=age, rss_bytes=rss_bytes, vms_bytes=vms_bytes, utime_ticks=stat_fields.utime_ticks, stime_ticks=stat_fields.stime_ticks, cpu_percent_interval=cpu_percent, num_threads=threads or stat_fields.num_threads, argv=argv, argv_display=argv_display, argv_masked_count=masked_count, exe_path=exe_path, cwd=cwd, cgroup=cgroup, read_errors=tuple(errors), ) def scan_processes( proc_root: str | Path = "/proc", *, previous: Mapping[str, ProcessSample] | None = None, current_uid_only: bool = True, exclude_pids: set[int] | None = None, sample_ts: float | None = None, ) -> list[ProcessSample]: root = Path(proc_root) previous = previous or {} exclude_pids = exclude_pids or set() sample_ts = time.time() if sample_ts is None else sample_ts boot_id = read_boot_id(root) uptime_seconds = read_uptime(root) samples: list[ProcessSample] = [] try: pid_dirs = list(root.iterdir()) except OSError: return samples for pid_dir in pid_dirs: if not pid_dir.name.isdigit(): continue sample = _sample_pid( pid_dir, boot_id=boot_id, uptime_seconds=uptime_seconds, sample_ts=sample_ts, previous=previous, current_uid_only=current_uid_only, exclude_pids=exclude_pids, ) if sample is not None: samples.append(sample) samples.sort(key=lambda sample: sample.pid) return samples def sample_process( pid: int, proc_root: str | Path = "/proc", *, previous: Mapping[str, ProcessSample] | None = None, current_uid_only: bool = True, exclude_pids: set[int] | None = None, sample_ts: float | None = None, ) -> ProcessSample | None: root = Path(proc_root) previous = previous or {} exclude_pids = exclude_pids or set() sample_ts = time.time() if sample_ts is None else sample_ts return _sample_pid( root / str(pid), boot_id=read_boot_id(root), uptime_seconds=read_uptime(root), sample_ts=sample_ts, previous=previous, current_uid_only=current_uid_only, exclude_pids=exclude_pids, )