internet_use_control/windows/webrestrict/process_watch.py

"""The Windows side of enforcement: process list, window list, foregound app.

Everything here touches the real machine, so it is kept apart from the pure
decision in :mod:`webrestrict.blocker`. ``psutil`` is the only third-party
dependency; the window queries go through ``ctypes`` and ``user32`` so the
client needs no ``pywin32`` install.

Off Windows (unit tests, a lab box) every function degrades to an empty result
instead of raising, so the client stays importable and testable.
"""

from __future__ import annotations

import ctypes
import logging
import os
import sys
from dataclasses import dataclass
from datetime import datetime
from typing import Iterable

from .blocker import Victim, select_victims, self_protection
from .config import DeviceConfig, bare

log = logging.getLogger("webrestrict.process")

IS_WINDOWS = os.name == "nt"

# How long to give a polite terminate() before the hard kill(), in seconds.
TERMINATE_GRACE_SECONDS = 3.0


def _psutil():
    """Import psutil on demand, with a readable error when it is missing."""
    try:
        import psutil  # noqa: PLC0415 (deliberate: keeps the import cost local)
    except ImportError as exc:  # pragma: no cover - depends on the machine
        raise RuntimeError(
            "psutil is not installed; run: python -m pip install -r windows/requirements.txt"
        ) from exc
    return psutil


def snapshot_processes() -> list[tuple[int, str]]:
    """Every running process as ``(pid, name)``.

    A process that exits while the list is built, or that denies access, is
    skipped: it is not a reason to fail an enforcement tick.
    """
    psutil = _psutil()
    out: list[tuple[int, str]] = []
    for proc in psutil.process_iter(["pid", "name"]):
        try:
            info = proc.info
            name = info.get("name") or ""
            pid = int(info.get("pid") or 0)
        except Exception:  # noqa: BLE001 - a dead process is normal
            continue
        if pid and name:
            out.append((pid, name))
    return out


def kill_process(pid: int, name: str) -> str:
    """Terminate one process. Returns ``"terminated"`` or ``"killed"``.

    ``terminate()`` asks the app to close, which is exactly what a scheduled
    window does to a game. If it is still alive after a short grace, ``kill()``
    ends it. Access denied is reported, not raised: the machine keeps running.
    """
    psutil = _psutil()
    try:
        proc = psutil.Process(pid)
    except Exception as exc:  # noqa: BLE001 - the process is already gone
        log.debug("process %s (%s) vanished: %s", pid, name, exc)
        return "gone"

    try:
        proc.terminate()
    except Exception as exc:  # noqa: BLE001 - e.g. AccessDenied for another user
        log.warning("terminate %s (%s) failed: %s", pid, name, exc)
        return "failed"

    try:
        proc.wait(timeout=TERMINATE_GRACE_SECONDS)
        return "terminated"
    except Exception:  # noqa: BLE001 - still running, escalate
        pass

    try:
        proc.kill()
        proc.wait(timeout=TERMINATE_GRACE_SECONDS)
        return "killed"
    except Exception as exc:  # noqa: BLE001
        log.warning("kill %s (%s) failed: %s", pid, name, exc)
        return "failed"


# --- Windows window probes (ctypes; no pywin32) --------------------------

_WNDENUMPROC = ctypes.WINFUNCTYPE(ctypes.c_bool, ctypes.c_void_p, ctypes.c_void_p) if IS_WINDOWS else None


def _user32():
    return ctypes.windll.user32  # type: ignore[attr-defined]


def window_pids() -> set[int]:
    """Pids that own a visible, titled top-level window.

    That is the closest cheap proxy Windows offers for "the child can see this
    app". A background app with a hidden window does not count.
    """
    if not IS_WINDOWS:  # pragma: no cover - platform dependent
        return set()

    user32 = _user32()
    pids: set[int] = set()

    def callback(hwnd, _lparam):  # pragma: no cover - platform dependent
        try:
            if not user32.IsWindowVisible(hwnd):
                return True
            if user32.GetWindowTextLengthW(hwnd) == 0:
                return True
            pid = ctypes.c_ulong(0)
            user32.GetWindowThreadProcessId(hwnd, ctypes.byref(pid))
            if pid.value:
                pids.add(int(pid.value))
        except Exception:  # noqa: BLE001 - never let a callback break the enum
            return True
        return True

    try:
        user32.EnumWindows(_WNDENUMPROC(callback), 0)
    except Exception as exc:  # noqa: BLE001 - pragma: no cover
        log.debug("EnumWindows failed: %s", exc)
    return pids


def foreground_pid() -> int | None:
    """The pid of the window the child is using, or ``None`` off Windows."""
    if not IS_WINDOWS:  # pragma: no cover - platform dependent
        return None
    try:
        hwnd = _user32().GetForegroundWindow()
        if not hwnd:
            return None
        pid = ctypes.c_ulong(0)
        _user32().GetWindowThreadProcessId(hwnd, ctypes.byref(pid))
        return int(pid.value) or None
    except Exception:  # noqa: BLE001 - pragma: no cover
        return None


def process_name(pid: int) -> str:
    """The image name for one pid, or an empty string."""
    psutil = _psutil()
    try:
        return psutil.Process(pid).name() or ""
    except Exception:  # noqa: BLE001 - the process ended
        return ""


def foreground_app() -> str:
    """The foreground app's image name, for the status report (FR-9)."""
    pid = foreground_pid()
    return process_name(pid) if pid else ""


def visible_apps(exclude_pids: Iterable[int] | None = None) -> list[str]:
    """Names of processes with a visible, titled window right now.

    The client's own process is excluded by default: its warning window has a
    title and would otherwise be counted as the child's usage.
    """
    pids = window_pids()
    if not pids:
        return []
    skip = {int(pid) for pid in (exclude_pids if exclude_pids is not None else (os.getpid(),))}
    pids -= skip
    return sorted({name for name in (process_name(pid) for pid in pids) if name})


@dataclass
class EnforcementReport:
    """What one tick did, for the log and the status report."""

    blocking: bool
    blocked_names: tuple[str, ...]
    killed: tuple[Victim, ...]
    failed: tuple[Victim, ...]

    @property
    def killed_names(self) -> list[str]:
        return sorted({victim.matched for victim in self.killed})


class Enforcer:
    """Terminates blocked processes while the device is effectively locked."""

    def __init__(self, protected: Iterable[str], log_extra: logging.Logger | None = None):
        self.log = log_extra or log
        self.protected_names = {bare(name) for name in protected if name and name.strip()}
        self.protected_pids: set[int] = set()
        # The client's own process tree is never a victim, even if a blocklist
        # names python.exe: killing itself would end enforcement silently.
        parents = _parent_pids(os.getpid())
        pids, names = self_protection(os.getpid(), {os.path.basename(sys.executable), "python", "pythonw"}, parents)
        self.protected_pids |= pids
        self.protected_names |= {bare(name) for name in names}

    def enforce(self, config: DeviceConfig, now: datetime | None = None) -> EnforcementReport:
        """One enforcement tick: find the victims, kill them, report the result."""
        blocking = config.is_blocking(now)
        victims = select_victims(
            snapshot_processes(),
            config,
            now=now,
            protected_names=self.protected_names,
            protected_pids=self.protected_pids,
        )
        killed: list[Victim] = []
        failed: list[Victim] = []
        for victim in victims:
            outcome = kill_process(victim.pid, victim.name)
            if outcome in ("terminated", "killed"):
                self.log.info(
                    "blocked %s (pid=%s, rule=%s): %s", victim.name, victim.pid, victim.matched, outcome
                )
                killed.append(victim)
            elif outcome == "failed":
                failed.append(victim)
        if killed:
            self.log.info("enforcement killed %d process(es): %s", len(killed), ", ".join(
                f"{v.name}({v.pid})" for v in killed
            ))
        return EnforcementReport(
            blocking=blocking,
            blocked_names=tuple(sorted(config.blocked_names())),
            killed=tuple(killed),
            failed=tuple(failed),
        )


def _parent_pids(pid: int) -> set[int]:
    """The pid's ancestor chain, skipping failures."""
    psutil = _psutil()
    out: set[int] = set()
    try:
        proc = psutil.Process(pid)
    except Exception:  # noqa: BLE001 - pragma: no cover
        return out
    for _ in range(10):
        try:
            parent = proc.parent()
        except Exception:  # noqa: BLE001
            break
        if parent is None or parent.pid in out:
            break
        out.add(parent.pid)
        proc = parent
    return out