diff --git a/gatelock/README.md b/gatelock/README.md index 52c179b..b643c5f 100644 --- a/gatelock/README.md +++ b/gatelock/README.md @@ -97,6 +97,25 @@ A claim from a pre-0.11 writer has no key and reads as `grab != "none"`. Root has no `XDG_RUNTIME_DIR`, so a root reader passes the user's dir (`/run/user/1000/gatelock/claims`). +`LockWindow` yields to a system credential prompt that holds the X grab. If the +global grab stays unobtainable for more than 2 s AND a mapped window whose +WM_CLASS is in `gatelock._prompt_names.SYSTEM_PROMPT_CLASSES` (`gcr-prompter`, +`pinentry*`, polkit agents, ssh-askpass variants) is on screen, that prompt's +top-level window is raised above the lock and the lock's periodic re-raise +stops covering it while it stays mapped. The lock stays mapped, VT switching +stays disabled and the grab keeps being retried; once the grab is taken the lock +is back on top. Any other grab holder (a fullscreen game) never gets this. +`scripts/verify_prompt_yield.py` proves it on a private Xvfb (never `:0`). + +Lockers that start together do not fight. `wait_for_turn` waits +`STARTUP_GRACE_SECONDS` (1.5 s) after `publish()` before its first check, so +simultaneous starters see each other. An armed lock also steps back (`_stepback`) +when a strictly higher-ranked, live claim that locks at least as firmly appears +later: it hides, releases the grab and its holder lock (the claim stays, so +lower ranks keep queueing), waits windowless, and re-arms fully once nothing +outranks it. `scripts/verify_all_lockers.py` checks this with five real +processes on a private Xvfb. + ## Development ```bash diff --git a/gatelock/gatelock/_arbiter.py b/gatelock/gatelock/_arbiter.py index 8f7c7ce..efc8b0b 100644 --- a/gatelock/gatelock/_arbiter.py +++ b/gatelock/gatelock/_arbiter.py @@ -222,14 +222,19 @@ class Arbiter: return self._claim return read_claim_if_held(path) + def drop_holder(self) -> None: + """Drop only the holder lock; the claim stays (see ``_stepback``).""" + if self._holder_handle is not None: + self._holder_handle.close() + self._holder_handle = None + def release(self) -> None: """Drop the holder lock and this app's claim. - Must run on every exit path. ``flock`` covers a killed process for - free, but a *cleanly* exiting app has to release explicitly -- the - morning routine runs the alarm and then the workout locker as - sequential subprocesses, and a lingering claim would make the workout - locker stand down against an app that had already finished. + Must run on every clean exit path (``flock`` covers a killed process): + the morning routine runs the alarm and then the workout locker as + sequential subprocesses, and a lingering claim would make the second + stand down against an app that had already finished. """ for handle in (self._holder_handle, self._claim_handle): if handle is not None: diff --git a/gatelock/gatelock/_arming.py b/gatelock/gatelock/_arming.py index d90c67a..289d5f8 100644 --- a/gatelock/gatelock/_arming.py +++ b/gatelock/gatelock/_arming.py @@ -22,11 +22,12 @@ Nothing here reaches back into the window, so the arming sequence can be read from __future__ import annotations import contextlib -from dataclasses import dataclass +from dataclasses import dataclass, field import logging import tkinter as tk from typing import TYPE_CHECKING +from gatelock._prompts import PromptYield from gatelock._surfaces import needs_backdrop_root from gatelock._vt import disable_vt_switching @@ -70,6 +71,8 @@ class ArmingCollaborators: still blocked, to name the holder and stand a weaker one down. notify_grab_held: Called once a grab call has actually succeeded, so the claim's ``grab_held`` turns true only then. + prompt_yield: Times the blocked grab (shared with the recovery loop) and + decides whether a system credential prompt gets raised over us. """ config: LockConfig @@ -80,6 +83,7 @@ class ArmingCollaborators: notify_focus_ready: Callable[[], None] log_grab_blocked: Callable[[int], None] notify_grab_held: Callable[[], None] = _no_grab_listener + prompt_yield: PromptYield = field(default_factory=PromptYield) def setup(root: tk.Tk, collab: ArmingCollaborators) -> bool: @@ -167,14 +171,16 @@ def acquire_global_grab( """ retry_ms = collab.config.grab.retry_ms try: + collab.prompt_yield.check_grab() root.grab_set_global() - except tk.TclError: + except tk.TclError as exc: if retry_ms == 0: - _logger.warning("Global grab failed, falling back to local grab") + _logger.warning("Global grab failed (%s), falling back to local grab", exc) with contextlib.suppress(tk.TclError): root.grab_set() collab.notify_grab_held() return + collab.prompt_yield.grab_blocked(str(exc)) effective_retry_ms = retry_ms or _DEFAULT_GRAB_RETRY_MS if not attempt % collab.config.grab.log_every: collab.log_grab_blocked(attempt) @@ -183,6 +189,8 @@ def acquire_global_grab( lambda: acquire_global_grab(root, collab, attempt=attempt + 1), ) return + if collab.prompt_yield.grab_held(): + collab.surfaces.raise_all() collab.notify_grab_held() with contextlib.suppress(tk.TclError): root.focus_force() diff --git a/gatelock/gatelock/_grabprobe.py b/gatelock/gatelock/_grabprobe.py new file mode 100644 index 0000000..e2a1c64 --- /dev/null +++ b/gatelock/gatelock/_grabprobe.py @@ -0,0 +1,132 @@ +# Copyright (c) 2026 Krzysztof Rudnicki +"""Asking X whether the global grab is free, without blocking Tk. + +Tk's ``grab -global`` retries internally and blocks the WHOLE event loop for +about 1.0 s on every failed attempt (measured 2026-11-11). The lock retries +every 200 ms and once per recovery tick, so while a foreign client held the +grab (the keyring prompt) the loop was starved: no repaint, no clicks, rows and +buttons without their text. + +The probe asks first, on a throwaway python-xlib connection: grab the keyboard +and the pointer on the root with ``GrabModeAsync`` and give them straight +back. ``AlreadyGrabbed`` means somebody else holds one, so the caller skips +Tk's call for this attempt; a failed request costs microseconds instead of a +second. Anything else -- free, or any status or error we do not understand -- +means "unknown", and the caller makes today's Tk call: the probe can only ever +*spare* a doomed attempt, never prevent one that might have worked. + +Deliberately not part of the ``_recovery_*`` family: it releases the grab it +briefly took, which that family's static invariant forbids by name. +""" + +from __future__ import annotations + +import contextlib +import logging +import tkinter as tk +from typing import TYPE_CHECKING, Protocol + +from gatelock._xlib import load_xlib, xlib_failures + +if TYPE_CHECKING: + from types import ModuleType + + # python-xlib ships no stubs; spell out the surface used (see _randr.py). + class _XRoot(Protocol): + def grab_keyboard(self, *args: int | bool) -> int: + """Mirror ``Window.grab_keyboard()``.""" + + def grab_pointer(self, *args: int | bool) -> int: + """Mirror ``Window.grab_pointer()``.""" + + class _XScreen(Protocol): + root: _XRoot + + class _XDisplay(Protocol): + def screen(self) -> _XScreen: + """Mirror ``Display.screen()``.""" + + def ungrab_keyboard(self, time: int) -> None: + """Mirror ``Display.ungrab_keyboard()``.""" + + def ungrab_pointer(self, time: int) -> None: + """Mirror ``Display.ungrab_pointer()``.""" + + def sync(self) -> None: + """Mirror ``Display.sync()``.""" + + +_logger = logging.getLogger(__name__) + +_OWNER_EVENTS = True +_NO_EVENTS = 0 + + +class GrabProbe: + """Answers "is another X client holding the keyboard or pointer?".""" + + def __init__(self) -> None: + """Start with nothing to repeat.""" + self._last_warning: str | None = None + + def _warn_once(self, message: str) -> None: + """Warn about a failure, but only when the message changes.""" + if message != self._last_warning: + _logger.warning("%s", message) + self._last_warning = message + + def check(self) -> None: + """Raise ``tk.TclError`` if the grab is known to be held elsewhere. + + Raised as Tk's own exception so the callers keep a single ``except`` + for "could not grab", whoever noticed. + """ + reason = self.blocked() + if reason is not None: + raise tk.TclError(reason) + + def blocked(self) -> str | None: + """Why the grab cannot be taken right now, or None if unknown or free.""" + modules = load_xlib("Xlib.display", "Xlib.X", "Xlib.error") + if modules is None: + self._warn_once("python-xlib is not installed; cannot probe the grab") + return None + xdisplay, xconst, xerror = modules + failures = xlib_failures(xerror) + try: + display = xdisplay.Display() + except failures as exc: + self._warn_once(f"cannot open an X connection to probe the grab: {exc}") + return None + try: + return _probe(display, xconst) + except failures as exc: + self._warn_once(f"probing the grab failed: {exc}") + return None + finally: + with contextlib.suppress(OSError): + display.close() + + +def _probe(display: _XDisplay, xconst: ModuleType) -> str | None: + """Grab keyboard then pointer on the root, release both, report a holder.""" + root = display.screen().root + modes = (xconst.GrabModeAsync, xconst.GrabModeAsync) + device = "keyboard" + status = root.grab_keyboard(_OWNER_EVENTS, *modes, xconst.CurrentTime) + if status == xconst.GrabSuccess: + device = "pointer" + status = root.grab_pointer( + _OWNER_EVENTS, + _NO_EVENTS, + *modes, + xconst.NONE, + xconst.NONE, + xconst.CurrentTime, + ) + display.ungrab_keyboard(xconst.CurrentTime) + display.ungrab_pointer(xconst.CurrentTime) + display.sync() + if status == xconst.AlreadyGrabbed: + return f"grab failed: another application has grab ({device}, probed)" + return None diff --git a/gatelock/gatelock/_placement.py b/gatelock/gatelock/_placement.py index 9bfdb22..81307d4 100644 --- a/gatelock/gatelock/_placement.py +++ b/gatelock/gatelock/_placement.py @@ -6,8 +6,9 @@ windows. This half owns the geometry questions -- is the window mapped, is it still exactly on its output -- and the re-assertion pass that answers "no" by correcting it. -Everything here only ever *adds* visibility. Nothing in this module can hide -or unmap a surface; that asymmetry is what makes the recovery loop monotonic. +Everything the recovery loop reaches here only ever *adds* visibility; that +asymmetry is what makes it monotonic. The one hider, :func:`set_visible`, is +reached only by :mod:`gatelock._stepback`. """ from __future__ import annotations @@ -55,7 +56,9 @@ def geometry_matches(surface: _Surface) -> bool: ) -def enforce_one(surface: _Surface, *, overrideredirect: bool) -> bool: +def enforce_one( + surface: _Surface, *, overrideredirect: bool, lift: bool = True +) -> bool: """Re-assert one surface's placement and stacking. Args: @@ -63,6 +66,9 @@ def enforce_one(surface: _Surface, *, overrideredirect: bool) -> bool: overrideredirect: Whether the lock runs WM-unmanaged, in which case the flag is re-set -- it can be dropped across an unmap/map cycle, and a managed lock window would be moved by the WM. + lift: Whether to raise the window. False only while a system prompt + is being yielded to (:mod:`gatelock._prompts`); placement and + mapping are corrected either way. Returns: True if anything needed correcting. @@ -77,10 +83,32 @@ def enforce_one(surface: _Surface, *, overrideredirect: bool) -> bool: changed = True if overrideredirect: window.overrideredirect(boolean=True) - window.lift() + if lift: + window.lift() return changed +def set_visible( + surfaces: Mapping[str, _Surface], *, visible: bool, overrideredirect: bool +) -> None: + """Hide or show every surface. Only a stepping-back lock ever hides. + + Showing restores the creation order (override-redirect while still + unmapped, geometry, then map and raise) rather than ``enforce_one``'s + map-first order, so a surface cannot come back window-manager-managed. + """ + for surface in surfaces.values(): + window = surface.window + if not visible: + window.withdraw() + continue + if overrideredirect: + window.overrideredirect(boolean=True) + set_geometry(window, surface.info.rect) + window.deiconify() + window.lift() + + def blind_outputs( surfaces: Mapping[str, _Surface], names: frozenset[str] ) -> tuple[str, ...]: @@ -105,13 +133,14 @@ def blind_outputs( def enforce_all( - surfaces: Mapping[str, _Surface], *, overrideredirect: bool + surfaces: Mapping[str, _Surface], *, overrideredirect: bool, lift: bool = True ) -> tuple[str, ...]: """Re-assert every surface's placement and stacking. Args: surfaces: The live surfaces, keyed by output name. overrideredirect: Whether the lock runs WM-unmanaged. + lift: Whether to raise the surfaces; see :func:`enforce_one`. Returns: The names of the outputs whose surfaces needed correcting, sorted. @@ -119,7 +148,7 @@ def enforce_all( corrected = [ surface.info.output_name for surface in surfaces.values() - if enforce_one(surface, overrideredirect=overrideredirect) + if enforce_one(surface, overrideredirect=overrideredirect, lift=lift) ] return tuple(sorted(corrected)) diff --git a/gatelock/gatelock/_preempt.py b/gatelock/gatelock/_preempt.py index 54e3c7f..da402e0 100644 --- a/gatelock/gatelock/_preempt.py +++ b/gatelock/gatelock/_preempt.py @@ -22,22 +22,39 @@ if TYPE_CHECKING: from gatelock._arbiter import Arbiter from gatelock._claims import Claim from gatelock._config import LockConfig + from gatelock._prompts import PromptYield _logger = logging.getLogger(__name__) +def _foreign_holder(arbiter: Arbiter | None) -> Claim | None: + """The gatelock app holding the screen, unless that app is this process. + + ``Arbiter.describe_holder`` answers "who owns the holder file", and when + this process wrote it that is our own claim. Reporting it as the grab's + holder made the lock blame itself (2026-11-11) and hid the real holder. + """ + holder = arbiter.describe_holder() if arbiter else None + if holder is None or arbiter is None: + return None + return None if holder.instance_id == arbiter.claim.instance_id else holder + + def log_grab_blocked( attempt: int, *, arbiter: Arbiter | None, config: LockConfig, preempted_pids: set[int], + prompt_yield: PromptYield | None = None, ) -> None: """Say who is actually holding the grab, rather than guessing. v0.1.1 always blamed "a fullscreen game". On 2026-07-25 the holder was the other locker, and that guess sent the diagnosis in the wrong - direction for the length of the outage. + direction for the length of the outage. On 2026-11-11 the lock blamed + *itself* (its own holder file) while the holder was the GNOME Keyring + prompt, so a holder that is this process counts as "no gatelock holder". A holder that outranks us is left alone entirely -- retrying is the correct behavior there, matching :func:`gatelock.wait_for_turn`'s own @@ -46,24 +63,36 @@ def log_grab_blocked( as long as it takes to finish on its own (diet_guard held the grab for ~3 minutes against a higher-ranked screen_locker on 2026-08-21). + X cannot say which client holds a grab, so credential prompts seen on + screen are named as *present*, not as the verdict. + Args: attempt: 1-based attempt counter, reported in the log line. arbiter: This app's arbiter, or None when arbitration is disabled. config: The window's configuration, consulted for the preempt opt-in. preempted_pids: Pids already signalled; mutated by the preempt path. + prompt_yield: The grab-block policy; supplies Tk's error text for the + failed grab and the system prompts seen on screen. """ - holder = arbiter.describe_holder() if arbiter else None + reason = prompt_yield.last_reason if prompt_yield else "" + prompts = prompt_yield.prompt_names if prompt_yield else () + cause = f" ({reason})" if reason else "" + holder = _foreign_holder(arbiter) if holder is None: _logger.warning( - "global grab still blocked after %d attempts; no gatelock app " - "holds it -- likely another X client (e.g. a fullscreen game)", + "global grab still blocked after %d attempts%s; no other gatelock " + "app holds it -- a non-gatelock X client does (e.g. a fullscreen " + "game or a credential prompt)%s", attempt, + cause, + f"; system prompt on screen: {', '.join(prompts)}" if prompts else "", ) return _logger.warning( - "global grab still blocked after %d attempts; held by gatelock app " + "global grab still blocked after %d attempts%s; held by gatelock app " "%r (rank %d, pid %d) since %s", attempt, + cause, holder.app, holder.rank, holder.pid, diff --git a/gatelock/gatelock/_prompt_names.py b/gatelock/gatelock/_prompt_names.py new file mode 100644 index 0000000..6138223 --- /dev/null +++ b/gatelock/gatelock/_prompt_names.py @@ -0,0 +1,62 @@ +# Copyright (c) 2026 Krzysztof Rudnicki +"""Which X windows are known system credential prompts. + +Incident 2026-11-11: at login the GNOME Keyring unlock prompt (``gcr-prompter``) +took the X keyboard and pointer grab ~1 s before a lock armed. The lock's +override-redirect windows then covered the prompt, so the grab could never be +taken and nothing on screen could be clicked. Yielding to *these* windows -- +and only these -- is the one deliberate exception to "the lock is always on +top"; anything else holding the grab (a fullscreen game) stays covered. +""" + +from __future__ import annotations + +from dataclasses import dataclass +from fnmatch import fnmatchcase +from typing import TYPE_CHECKING, Final + +if TYPE_CHECKING: + from collections.abc import Iterable + +SYSTEM_PROMPT_CLASSES: Final = ( + "gcr-prompter", + "pinentry*", + "polkit-gnome-authentication-agent-1", + "lxpolkit", + "ssh-askpass*", + "x11-ssh-askpass", + "ksshaskpass", + "gnome-ssh-askpass", +) +"""``fnmatch`` patterns, lowercase, matched against WM_CLASS instance or class.""" + + +@dataclass(frozen=True) +class PromptWindow: + """A mapped system prompt: the client and the ancestor below the root. + + Attributes: + window_id: The client window that carries the WM_CLASS. + toplevel_id: Its ancestor that is a direct child of the root. Under a + reparenting window manager (i3) that is the WM's frame, and it is + the window whose stacking position matters. + wm_class: The ``(instance, class)`` pair. + """ + + window_id: int + toplevel_id: int + wm_class: tuple[str, str] + + @property + def name(self) -> str: + """The WM_CLASS instance, for log lines.""" + return self.wm_class[0] + + +def is_system_prompt(wm_class: Iterable[str]) -> bool: + """Whether a WM_CLASS (instance or class, any case) is an allowlisted prompt.""" + return any( + fnmatchcase(part.lower(), pattern) + for part in wm_class + for pattern in SYSTEM_PROMPT_CLASSES + ) diff --git a/gatelock/gatelock/_prompts.py b/gatelock/gatelock/_prompts.py new file mode 100644 index 0000000..9997f13 --- /dev/null +++ b/gatelock/gatelock/_prompts.py @@ -0,0 +1,151 @@ +# Copyright (c) 2026 Krzysztof Rudnicki +"""Stepping aside for a system credential prompt that holds the X grab. + +Incident 2026-11-11: the GNOME Keyring prompt (``gcr-prompter``) was mapped and +holding the X keyboard+pointer grab when a lock armed. The lock's +override-redirect windows covered it, so the grab could never be taken, every +click went to a window nobody could see, and the lock's own UI was dead. + +The policy (approved by the user): when the global grab has been unobtainable +for more than :data:`YIELD_AFTER_S` AND a mapped window on the allowlist +(:data:`gatelock._prompt_names.SYSTEM_PROMPT_CLASSES`) is on screen, raise that +prompt above the lock and stop the lock's own periodic re-raise from covering +it again, for as long as the prompt stays mapped. Everything else stays as it +was: the lock is still mapped everywhere, VT switching stays disabled and the +grab is still retried every tick. Once the grab is taken the lock stacks on top +again. Anything not on the allowlist -- a fullscreen game -- is never yielded +to. + +Known limitation of an allowlist keyed on WM_CLASS: a window that merely +*claims* one of those classes while holding a grab would be raised too. + +This module is the policy and the clock. All X access is in +:mod:`gatelock._prompts_x`. It is deliberately NOT named ``_recovery_*``: the +recovery family is statically checked to contain no release call, and keeping +the X plumbing out of it keeps that check meaningful. +""" + +from __future__ import annotations + +import logging +import time +from typing import TYPE_CHECKING, Final, Protocol + +from gatelock._grabprobe import GrabProbe +from gatelock._prompts_x import XPromptProbe + +if TYPE_CHECKING: + from collections.abc import Callable + + from gatelock._prompt_names import PromptWindow + +_logger = logging.getLogger(__name__) + +YIELD_AFTER_S: Final = 2.0 +"""How long the grab must stay unobtainable before looking for a prompt.""" + + +class PromptProbe(Protocol): + """What the policy needs from the X side.""" + + def raise_prompts(self) -> tuple[PromptWindow, ...]: + """Raise every mapped allowlisted prompt; return those it raised.""" + + +class PromptYield: + """Times a blocked grab and decides whether to raise a prompt over the lock.""" + + def __init__( + self, + probe: PromptProbe | None = None, + *, + grab_probe: GrabProbe | None = None, + clock: Callable[[], float] = time.monotonic, + after_s: float = YIELD_AFTER_S, + ) -> None: + """Build the policy; nothing is blocked yet. + + Args: + probe: Finds and raises prompts. Defaults to the python-xlib one. + grab_probe: Asks X whether the grab is free. Defaults to the real one. + clock: Monotonic seconds; injected by tests. + after_s: Seconds the grab must stay blocked before yielding. + """ + self._probe: PromptProbe = probe if probe is not None else XPromptProbe() + self._grab_probe = grab_probe if grab_probe is not None else GrabProbe() + self._clock = clock + self._after_s = after_s + self._blocked_since: float | None = None + self._reason = "" + self._prompts: tuple[PromptWindow, ...] = () + + @property + def last_reason(self) -> str: + """Tk's message for the most recent failed grab (empty if none yet).""" + return self._reason + + @property + def yielding(self) -> bool: + """Whether a prompt is being yielded to right now.""" + return bool(self._prompts) + + @property + def prompt_names(self) -> tuple[str, ...]: + """WM_CLASS instances of the prompts currently yielded to.""" + return tuple(prompt.name for prompt in self._prompts) + + def check_grab(self) -> None: + """Raise ``tk.TclError`` if the grab is known to be held elsewhere. + + Call before every ``grab_set_global()``: Tk blocks its whole event + loop for ~1 s on each failed attempt, so a doomed one is skipped. + """ + self._grab_probe.check() + + def grab_blocked(self, reason: str) -> None: + """Record a failed grab; the first failure starts the clock. + + Args: + reason: Tk's error text ("another application has grab", ...). + """ + self._reason = reason + if self._blocked_since is None: + self._blocked_since = self._clock() + + def grab_held(self) -> bool: + """Record that the grab is ours; end any yield. + + Returns: + True when a yield was in progress, so the caller must lift the + lock over whatever it had stepped aside for. + """ + self._blocked_since = None + was_yielding = self.yielding + self._prompts = () + if was_yielding: + _logger.warning("global grab taken; the lock stacks on top again") + return was_yielding + + def settle(self) -> None: + """Re-decide, and re-raise the prompt. Call LAST in a recovery tick. + + Last, so that nothing the tick lifted can end up above the prompt. + """ + since = self._blocked_since + if since is None or self._clock() - since < self._after_s: + return + previous = self.prompt_names + self._prompts = self._probe.raise_prompts() + if self.prompt_names == previous: + return + if self._prompts: + _logger.warning( + "global grab blocked for %.1fs (%s) and system prompt %s is on " + "screen; raising it above the lock so it can be answered. The " + "lock stays up, VT stays disabled, the grab keeps being retried", + self._clock() - since, + self._reason, + ", ".join(self.prompt_names), + ) + else: + _logger.info("system prompt %s is gone; lock on top", ", ".join(previous)) diff --git a/gatelock/gatelock/_prompts_x.py b/gatelock/gatelock/_prompts_x.py new file mode 100644 index 0000000..31e68c3 --- /dev/null +++ b/gatelock/gatelock/_prompts_x.py @@ -0,0 +1,176 @@ +# Copyright (c) 2026 Krzysztof Rudnicki +"""Finding a system credential prompt on screen and raising it, over python-xlib. + +Kept apart from the policy in :mod:`gatelock._prompts` so that everything that +touches the X server lives in one place, behind :class:`XPromptProbe`. It opens +a short-lived connection per call: the caller asks at most once a second, and +only while the grab has been blocked for a while, so a persistent connection +would need a teardown path for no gain. + +Candidates come from the window manager's ``_NET_CLIENT_LIST`` *and* a walk of +the root's children and grandchildren. The walk is not redundant: a bare X +server has no ``_NET_CLIENT_LIST``, and under i3 the WM_CLASS sits on the +client, one level below the frame i3 reparents it into. What gets raised is +the ancestor directly below the root (the frame), never the client. + +Every X failure -- no python-xlib, no connection, a window that vanished +between listing and reading it (BadWindow) -- ends the scan with "no prompt +found", warned once per distinct message: the lock must stay on top whenever +it cannot be sure. +""" + +from __future__ import annotations + +import contextlib +import logging +from typing import TYPE_CHECKING, Protocol + +from gatelock._prompt_names import PromptWindow, is_system_prompt +from gatelock._xlib import load_xlib, xlib_failures + +if TYPE_CHECKING: + from collections.abc import Callable, Sequence + from types import ModuleType + + # python-xlib ships no stubs; spell out the surface used (see _randr.py). + class _Tree(Protocol): + children: Sequence[_XWindow] + parent: _XWindow + + class _Attributes(Protocol): + map_state: int + + class _Property(Protocol): + value: Sequence[int] + + class _XWindow(Protocol): + id: int + + def query_tree(self) -> _Tree: + """Mirror ``Window.query_tree()``.""" + + def get_wm_class(self) -> tuple[str, str] | None: + """Mirror ``Window.get_wm_class()``.""" + + def get_attributes(self) -> _Attributes: + """Mirror ``Window.get_attributes()``.""" + + def get_full_property(self, atom: object, kind: int) -> _Property | None: + """Mirror ``Window.get_full_property()``.""" + + def configure( + self, *, stack_mode: int, onerror: Callable[[object], None] + ) -> None: + """Mirror ``Window.configure()``.""" + + class _XScreen(Protocol): + root: _XWindow + + class _XDisplay(Protocol): + def screen(self) -> _XScreen: + """Mirror ``Display.screen()``.""" + + def intern_atom(self, name: str) -> object: + """Mirror ``Display.intern_atom()``.""" + + def create_resource_object(self, kind: str, window_id: int) -> _XWindow: + """Mirror ``Display.create_resource_object()``.""" + + def sync(self) -> None: + """Mirror ``Display.sync()``.""" + + def close(self) -> None: + """Mirror ``Display.close()``.""" + + +_logger = logging.getLogger(__name__) + + +class XPromptProbe: + """Looks for allowlisted prompts and raises their top-level windows.""" + + def __init__(self) -> None: + """Start with nothing to repeat.""" + self._last_warning: str | None = None + + def _warn_once(self, message: str) -> None: + """Warn about a failure, but only when the message changes.""" + if message != self._last_warning: + _logger.warning("%s", message) + self._last_warning = message + + def raise_prompts(self) -> tuple[PromptWindow, ...]: + """Raise every mapped system prompt above everything else. + + Returns: + The prompts that are on screen and were raised; empty when there + are none, or when anything went wrong (the lock then stays on top). + """ + modules = load_xlib("Xlib.display", "Xlib.X", "Xlib.error") + if modules is None: + self._warn_once("python-xlib is not installed; cannot look for a prompt") + return () + xdisplay, xconst, xerror = modules + failures = xlib_failures(xerror) + try: + display: _XDisplay = xdisplay.Display() + except failures as exc: + self._warn_once(f"cannot open an X connection to look for a prompt: {exc}") + return () + refused: list[object] = [] + try: + found = _find(display, xconst) + for prompt in found: + top = display.create_resource_object("window", prompt.toplevel_id) + top.configure(stack_mode=xconst.Above, onerror=refused.append) + display.sync() + except failures as exc: + self._warn_once(f"looking for a prompt failed: {exc}") + return () + finally: + with contextlib.suppress(OSError): + display.close() + if refused: + self._warn_once(f"the X server refused to raise the prompt: {refused[0]}") + return () + self._last_warning = None + return found + + +def _find(display: _XDisplay, xconst: ModuleType) -> tuple[PromptWindow, ...]: + """Every viewable allowlisted window, with its top-level ancestor.""" + root = display.screen().root + ids: dict[int, None] = {} + listed = root.get_full_property( + display.intern_atom("_NET_CLIENT_LIST"), xconst.AnyPropertyType + ) + if listed is not None: + ids.update(dict.fromkeys(int(value) for value in listed.value)) + for child in root.query_tree().children: + ids[child.id] = None + ids.update(dict.fromkeys(grand.id for grand in child.query_tree().children)) + found: list[PromptWindow] = [] + for window_id in ids: + window = display.create_resource_object("window", window_id) + wm_class = window.get_wm_class() + if not wm_class or not is_system_prompt(wm_class): + continue + if window.get_attributes().map_state != xconst.IsViewable: + continue + found.append( + PromptWindow( + window_id=window_id, + toplevel_id=_toplevel(window, root.id), + wm_class=(str(wm_class[0]), str(wm_class[1])), + ) + ) + return tuple(found) + + +def _toplevel(window: _XWindow, root_id: int) -> int: + """The id of ``window``'s ancestor that is a direct child of the root.""" + top = window + parent = top.query_tree().parent + while parent.id != root_id: + top, parent = parent, parent.query_tree().parent + return top.id diff --git a/gatelock/gatelock/_queue.py b/gatelock/gatelock/_queue.py index 3b450e4..4ed1871 100644 --- a/gatelock/gatelock/_queue.py +++ b/gatelock/gatelock/_queue.py @@ -30,6 +30,12 @@ from typing import TYPE_CHECKING, Final _logger: Final = logging.getLogger(__name__) QUEUE_POLL_SECONDS: Final = 2.0 +# After publishing, wait this long before the FIRST turn check, so lockers that +# start together see each other's claims. Without it a weaker locker whose +# check ran before a stronger one had published armed alone, and the stronger +# then armed on top: two windows, the weaker holding the grab, the visible one +# unclickable. Read at call time (tests set it to 0). +STARTUP_GRACE_SECONDS: Final = 1.5 QUEUE_DEADLINE_SECONDS: Final = 6 * 60 * 60 # How often to re-state a still-blocked wait at WARNING. A long wait used # to produce exactly one INFO line at the moment it started and nothing @@ -206,6 +212,8 @@ def wait_for_turn( started = clock() wait = _Wait(arbiter, on_state, started) + if STARTUP_GRACE_SECONDS > 0: + time_source.sleep(STARTUP_GRACE_SECONDS) while True: blockers = stronger_claims(arbiter) diff --git a/gatelock/gatelock/_recovery.py b/gatelock/gatelock/_recovery.py index f1cec80..b2c2811 100644 --- a/gatelock/gatelock/_recovery.py +++ b/gatelock/gatelock/_recovery.py @@ -41,6 +41,7 @@ import logging import tkinter as tk from gatelock._grabwatch import GrabWatch +from gatelock._recovery_grab import reassert_grab from gatelock._recovery_types import RecoveryCollaborators, RecoveryReport from gatelock._vt import disable_vt_switching @@ -63,6 +64,11 @@ class RecoveryLoop: self._verify_job: str | None = None self._grab = GrabWatch(root) + @property + def running(self) -> bool: + """Whether the loop is enforcing (False once stopped, e.g. study mode).""" + return self._running + @property def ticks(self) -> int: """How many full verify passes have run.""" @@ -146,13 +152,17 @@ class RecoveryLoop: ) return RecoveryReport(scan_ok=False, source=scan.source) + # Read BEFORE enforce: while a system prompt is being yielded to, the + # tick must not lift the lock over it. The prompt is re-raised LAST. + yielding = self._parts.prompt_yield.yielding self._parts.surfaces.update_backdrop() delta = self._parts.surfaces.apply(scan) - corrected = self._parts.surfaces.enforce() + corrected = self._parts.surfaces.enforce(lift=not yielding) grab_reasserted = self._reassert_grab() vt_reasserted = self._reassert_vt() if delta.created: self._reassert_focus() + self._parts.prompt_yield.settle() if delta.unverified: _logger.error( @@ -188,20 +198,9 @@ class RecoveryLoop: def _reassert_grab(self) -> bool: """Re-take the global grab if we lost it. Never releases one. - Returns: - True if the grab had to be re-taken. + Delegates to :func:`gatelock._recovery_grab.reassert_grab`. """ - if self._parts.config.resolved_grab() != "global": - return False - held = self.holds_grab() - reasserted = False - if not held: - with contextlib.suppress(tk.TclError): - self._root.grab_set_global() - _logger.warning("global grab had been lost; re-acquired it") - reasserted = held = True - self._parts.on_grab_held(held=held) - return reasserted + return reassert_grab(self._root, self._parts, self._grab) def holds_grab(self) -> bool: """Whether the grab is held by a window this application owns. diff --git a/gatelock/gatelock/_recovery_grab.py b/gatelock/gatelock/_recovery_grab.py new file mode 100644 index 0000000..ea09cb8 --- /dev/null +++ b/gatelock/gatelock/_recovery_grab.py @@ -0,0 +1,57 @@ +# Copyright (c) 2026 Krzysztof Rudnicki +"""Re-taking a lost global grab, once per recovery tick. + +Split out of :mod:`gatelock._recovery` for the 250-line cap, and kept in its +``_recovery_*`` family on purpose: the static invariant test globs the family, +so this file is held to the same rule -- it may take the grab, never release it. +""" + +from __future__ import annotations + +import logging +import tkinter as tk +from typing import TYPE_CHECKING + +if TYPE_CHECKING: + from gatelock._grabwatch import GrabWatch + from gatelock._recovery_types import RecoveryCollaborators + +_logger = logging.getLogger(__name__) + + +def reassert_grab( + root: tk.Misc, parts: RecoveryCollaborators, watch: GrabWatch +) -> bool: + """Re-take the global grab if we lost it. Never releases one. + + A failed attempt is reported with Tk's own message (once per distinct + message: this runs every tick) and timed by the prompt-yield policy. A + grab we hold ends any yield, and then the lock is lifted over whatever it + had stepped aside for. + + Args: + root: The Tk root that holds the grab. + parts: The loop's collaborators. + watch: Answers "is the grab ours?" and throttles the warning. + + Returns: + True if the grab had to be re-taken. + """ + if parts.config.resolved_grab() != "global": + return False + held = watch.holds_grab() + reasserted = False + if not held: + try: + parts.prompt_yield.check_grab() + root.grab_set_global() + except tk.TclError as exc: + parts.prompt_yield.grab_blocked(str(exc)) + watch.warn_once(f"could not re-take the global grab ({exc})") + else: + _logger.warning("global grab had been lost; re-acquired it") + reasserted = held = True + if held and parts.prompt_yield.grab_held(): + parts.surfaces.raise_all() + parts.on_grab_held(held=held) + return reasserted diff --git a/gatelock/gatelock/_recovery_types.py b/gatelock/gatelock/_recovery_types.py index 53da1de..5811809 100644 --- a/gatelock/gatelock/_recovery_types.py +++ b/gatelock/gatelock/_recovery_types.py @@ -8,9 +8,11 @@ against a report without pulling in the machinery that builds one. from __future__ import annotations -from dataclasses import dataclass +from dataclasses import dataclass, field from typing import TYPE_CHECKING, Protocol +from gatelock._prompts import PromptYield + if TYPE_CHECKING: from gatelock._config import LockConfig from gatelock._detect import OutputChangeDetector @@ -60,3 +62,5 @@ class RecoveryCollaborators: hooks: LockWindowHooks on_grab_held: GrabHeldListener = ignore_grab_held """Told each tick whether the grab is really ours (``Claim.grab_held``).""" + prompt_yield: PromptYield = field(default_factory=PromptYield) + """Times a blocked grab and raises a system prompt over the lock.""" diff --git a/gatelock/gatelock/_stepback.py b/gatelock/gatelock/_stepback.py new file mode 100644 index 0000000..3c08d4f --- /dev/null +++ b/gatelock/gatelock/_stepback.py @@ -0,0 +1,186 @@ +# Copyright (c) 2026 Krzysztof Rudnicki +"""A weaker armed lock steps back for a stronger one that appears. + +Lockers that start together can still race: a weaker one that armed before a +stronger one had published stays armed, and the stronger then arms on top. Two +windows, the weaker one holding the grab, the visible one unclickable -- the +same symptom as the 2026-11-11 keyring incident. The startup grace in +:mod:`gatelock._queue` closes most of that window; this closes the rest (a +stronger locker that starts later, e.g. three seconds after). + +While armed, the lock checks about once a second for a **live** claim that is +**strictly** higher ranked AND locks at least as firmly (grab kind and VT: +the arbiter's own rule that rank must never weaken a lock). Equal, lower, +weaker-locking and dead claims never trigger it. On a hit the lock suspends: +recovery and detector stop, the grab is released, surfaces and backdrop are +hidden and the holder lock is dropped. The **claim stays published**, so +lower-ranked apps keep queueing behind it. VT switching is left as it is: the +stronger lock disables it itself, and re-disabling is part of re-arming. It then +waits windowless -- the same turn logic as :func:`gatelock.wait_for_turn` -- +and re-arms fully (VT, backdrop, surfaces, grab) once nothing qualifying is +live. It is the same process and lock: the app's state and callbacks are +untouched. + +This module holds the library's ONLY ``grab_release`` and the only caller of +``SurfaceSet.set_visible(visible=False)``; a test pins both. It is kept out of +the ``_recovery*`` family, whose static invariant forbids release calls. +""" + +from __future__ import annotations + +import logging +import time +import tkinter as tk +from typing import TYPE_CHECKING + +from gatelock._queue import QUEUE_DEADLINE_SECONDS, stronger_claims +from gatelock._vt import disable_vt_switching + +if TYPE_CHECKING: + from collections.abc import Callable + + from gatelock._arbiter import Arbiter + from gatelock._arming import ArmingCollaborators + from gatelock._claims import Claim + +_logger = logging.getLogger(__name__) + + +def outranking(arbiter: Arbiter) -> tuple[Claim, ...]: + """Live claims that must make us step back: higher rank, at least as firm. + + ``stronger_claims`` proves liveness with ``flock`` (a dead owner's claim is + reaped, never returned) and compares rank alone; the extra strength test is + what stops a stronger-ranked but weaker-locking app from disarming us. + """ + mine = arbiter.claim + return tuple(c for c in stronger_claims(arbiter) if c.at_least_as_strong_as(mine)) + + +def _attempt(action: Callable[[], None], what: str) -> None: + """Run one Tk step; a failure is logged and the rest still happens.""" + try: + action() + except tk.TclError as exc: + _logger.warning("step-back: %s failed (%s)", what, exc) + + +class StepBack: + """Suspends an armed lock for a stronger one, and re-arms it afterwards.""" + + def __init__( + self, + root: tk.Tk, + arming: ArmingCollaborators, + arbiter: Arbiter | None, + rearm: Callable[[], None], + *, + clock: Callable[[], float] = time.monotonic, + ) -> None: + """Wire the controller; nothing is scheduled until :meth:`start`. + + Args: + root: The Tk root (backdrop and grab holder). + arming: The window's arming collaborators (surfaces, loop, ...). + arbiter: This app's arbiter; None disables stepping back. + rearm: Re-takes the grab and restarts the loops + (``LockWindow.grab_input``). + clock: Monotonic seconds; injected by tests. + """ + self._root = root + self._arming = arming + self._arbiter = arbiter + self._rearm = rearm + self._clock = clock + self._job: str | None = None + self._since: float | None = None + + @property + def suspended(self) -> bool: + """Whether the lock is currently stepped back.""" + return self._since is not None + + def start(self) -> None: + """Begin (or restart) the periodic check. Idempotent.""" + if self._arbiter is not None: + self._schedule() + + def stop(self) -> None: + """Cancel the periodic check.""" + if self._job is not None: + _attempt(lambda: self._root.after_cancel(str(self._job)), "cancel") + self._job = None + + def _schedule(self) -> None: + """Queue the next check, replacing any pending one.""" + self.stop() + self._job = self._root.after(self._arming.config.recovery_tick_ms, self._tick) + + def _tick(self) -> None: + """One check; a raise must not end the checking.""" + self._job = None + try: + if self._since is None: + self._maybe_suspend() + else: + self._maybe_resume(self._since) + except Exception: # pylint: disable=broad-exception-caught + _logger.exception("step-back check raised; checking continues") + finally: + self._schedule() + + def _maybe_suspend(self) -> None: + """Step back if, and only if, a qualifying claim is live.""" + arbiter = self._arbiter + # A stopped loop means the app took the lock down on purpose (study + # mode); it is not ours to suspend or, later, to re-arm. + if arbiter is None or not self._arming.recovery.running: + return + blockers = outranking(arbiter) + if not blockers: + return + _logger.warning( + "%s (rank %d) steps back: %s is live and outranks it; hiding, " + "releasing the grab, waiting with no window", + arbiter.claim.app, + arbiter.claim.rank, + ", ".join(f"{c.app} (rank {c.rank})" for c in blockers), + ) + # First: if any step below raised, the next tick must still see us as + # suspended and re-arm, never as armed with a half-stopped loop. + self._since = self._clock() + self._arming.recovery.stop() + self._arming.detector.stop() + _attempt(self._root.grab_release, "grab release") + _attempt(lambda: self._arming.surfaces.set_visible(visible=False), "hide") + _attempt(self._root.withdraw, "backdrop hide") + arbiter.set_grab_held(held=False) + arbiter.drop_holder() + + def _maybe_resume(self, since: float) -> None: + """Re-arm once nothing qualifying is live (or the backstop is hit).""" + arbiter = self._arbiter + if arbiter is None: # pragma: no cover - never suspended without one + return + waited = self._clock() - since + if outranking(arbiter): + if waited < QUEUE_DEADLINE_SECONDS: + return + _logger.error( + "%s waited %.0fs stepped back and gave up -- re-arming anyway " + "rather than leaving the machine unlocked", + arbiter.claim.app, + waited, + ) + _logger.warning( + "%s re-arms after %.0fs stepped back: nothing outranks it now", + arbiter.claim.app, + waited, + ) + self._since = None + arbiter.acquire_holder() + if self._arming.config.resolved_disable_vt(): + disable_vt_switching() + _attempt(self._root.deiconify, "backdrop show") + _attempt(lambda: self._arming.surfaces.set_visible(visible=True), "show") + self._rearm() diff --git a/gatelock/gatelock/_surfaces.py b/gatelock/gatelock/_surfaces.py index d7ac6c7..e923590 100644 --- a/gatelock/gatelock/_surfaces.py +++ b/gatelock/gatelock/_surfaces.py @@ -27,10 +27,8 @@ recording, because each rules out an approach that looks reasonable: afterwards and the placement is ignored. 3. **A window-manager-managed toplevel cannot be placed per-output** -- i3 rewrites the geometry wholesale. Per-output placement *requires* - override-redirect, so ``attributes(fullscreen=True)`` is gone from the lock - path entirely: it is an EWMH request that an override-redirect window is - invisible to, and where it does apply it targets one monitor or the whole - bounding box. Neither is what per-output locking needs. + override-redirect, so ``attributes(fullscreen=True)`` is gone: an EWMH + request an override-redirect window never sees. """ from __future__ import annotations @@ -45,6 +43,7 @@ from gatelock._placement import ( enforce_all, needs_backdrop_root, set_geometry, + set_visible, ) from gatelock._surface_types import ( SurfaceBuilder, @@ -54,9 +53,7 @@ from gatelock._surface_types import ( from gatelock._surfaces_focus import focus_surface, preferred_focus_index from gatelock._textmirror import TextMirror, mirror_text_widgets -# Names this module re-exports after the 250-line split. Listed explicitly so -# the autofixer cannot prune an import that exists *for* its re-export -- it -# stripped mirror_text_widgets once, and `gatelock` itself stopped importing. +# Re-exports after the 250-line split; listed so the autofixer keeps them. __all__ = [ "SurfaceBuilder", "SurfaceDelta", @@ -161,27 +158,36 @@ class SurfaceSet: ) def verify(self, *, live_names: frozenset[str] | None = None) -> tuple[str, ...]: - """Return live outputs that are NOT currently showing a lock. - - This is the alarm for the original defect: an output the user can see, - with no lock on it, means the escape UI is somewhere they cannot reach. - """ + """Return live outputs that are NOT currently showing a lock.""" names = live_names if live_names is not None else self.names() return blind_outputs(self._surfaces, names) - def enforce(self) -> tuple[str, ...]: + def enforce(self, *, lift: bool = True) -> tuple[str, ...]: """Re-assert every surface's placement and stacking. - Only ever *adds* visibility: remaps what became unmapped, corrects - drifted geometry, and raises. Nothing here can hide a surface. + Only adds visibility: remaps, corrects drifted geometry, raises. + + Args: + lift: False while a system prompt is yielded to (no raise). """ return enforce_all( self._surfaces, overrideredirect=self._config.resolved_overrideredirect(), + lift=lift, + ) + + def set_visible(self, *, visible: bool) -> None: + """Hide or show every surface; only ``_stepback`` hides.""" + set_visible( + self._surfaces, + visible=visible, + overrideredirect=self._config.resolved_overrideredirect(), ) def raise_all(self) -> None: - """Lift every surface above whatever else is on screen.""" + """Lift the backdrop and every surface above whatever else is on screen.""" + if needs_backdrop_root(self._config): + self._root.lift() for surface in self._surfaces.values(): surface.window.lift() diff --git a/gatelock/gatelock/_window.py b/gatelock/gatelock/_window.py index 22368ab..a7359d6 100644 --- a/gatelock/gatelock/_window.py +++ b/gatelock/gatelock/_window.py @@ -31,7 +31,9 @@ from typing import TYPE_CHECKING from gatelock import _arming, _preempt from gatelock._detect import OutputChangeDetector from gatelock._outputs import OutputEnumerator +from gatelock._prompts import PromptYield from gatelock._recovery import RecoveryCollaborators, RecoveryLoop +from gatelock._stepback import StepBack from gatelock._surfaces import SurfaceSet from gatelock._vt import restore_vt_switching @@ -80,6 +82,7 @@ class LockWindow: self._closed = False self._focus_notified = False self._preempted_pids: set[int] = set() + prompt_yield = PromptYield() surfaces = SurfaceSet(root, config, hooks) enumerator = OutputEnumerator(root) detector = OutputChangeDetector(root) @@ -92,6 +95,7 @@ class LockWindow: detector=detector, hooks=hooks, on_grab_held=self.report_grab_held, + prompt_yield=prompt_yield, ), ) # The arming bundle is also where the window reaches its @@ -105,7 +109,9 @@ class LockWindow: notify_focus_ready=self._notify_focus_ready, log_grab_blocked=self._log_grab_blocked, notify_grab_held=lambda: self.report_grab_held(held=True), + prompt_yield=prompt_yield, ) + self._stepback = StepBack(root, self._arming, arbiter, self.grab_input) @property def surfaces(self) -> SurfaceSet: @@ -140,24 +146,16 @@ class LockWindow: # -- window mechanics ----------------------------------------------------- def setup(self) -> None: - """Configure the backdrop and build a surface on every live output. - - Delegates to :func:`gatelock._arming.setup`. - """ + """Build the backdrop and a surface per live output (``_arming.setup``).""" self._vt_disabled = _arming.setup(self.root, self._arming) def grab_input(self) -> None: - """Take the configured grab, then start watching for output changes. - - Delegates to :func:`gatelock._arming.grab_input`. - """ + """Take the configured grab, then watch outputs (``_arming.grab_input``).""" _arming.grab_input(self.root, self._arming) + self._stepback.start() def _acquire_global_grab(self, *, attempt: int) -> None: - """Acquire the global grab, retrying per ``grab_retry_ms``. - - Delegates to :func:`gatelock._arming.acquire_global_grab`. - """ + """Acquire the global grab, retrying (``_arming.acquire_global_grab``).""" _arming.acquire_global_grab(self.root, self._arming, attempt=attempt) def _log_grab_blocked(self, attempt: int) -> None: @@ -171,6 +169,7 @@ class LockWindow: arbiter=self._arbiter, config=self._config, preempted_pids=self._preempted_pids, + prompt_yield=self._arming.prompt_yield, ) def _notify_focus_ready(self) -> None: @@ -228,6 +227,7 @@ class LockWindow: if self._closed: return self._closed = True + self._stepback.stop() self._arming.recovery.stop() self._arming.detector.stop() self._hooks.on_close() diff --git a/gatelock/gatelock/_xlib.py b/gatelock/gatelock/_xlib.py index 85ffc31..3b453ab 100644 --- a/gatelock/gatelock/_xlib.py +++ b/gatelock/gatelock/_xlib.py @@ -31,3 +31,22 @@ def load_xlib(*names: str) -> tuple[ModuleType, ...] | None: return tuple(importlib.import_module(name) for name in names) except ImportError: return None + + +def xlib_failures(xerror: ModuleType) -> tuple[type[BaseException], ...]: + """Every exception a python-xlib connection or request can raise. + + Args: + xerror: The loaded ``Xlib.error`` module. + + Returns: + A tuple usable in an ``except`` clause. + """ + return ( + OSError, + ValueError, + xerror.DisplayError, + xerror.XauthError, + xerror.ConnectionClosedError, + xerror.XError, + ) diff --git a/gatelock/gatelock/tests/conftest.py b/gatelock/gatelock/tests/conftest.py index b097e7d..e766a45 100644 --- a/gatelock/gatelock/tests/conftest.py +++ b/gatelock/gatelock/tests/conftest.py @@ -17,6 +17,7 @@ from unittest.mock import MagicMock, patch import pytest +from gatelock._arbiter import Arbiter from gatelock._config import LockConfig from gatelock._outputs import Output, OutputRect, OutputScan from gatelock._recovery import RecoveryCollaborators, RecoveryLoop @@ -78,6 +79,12 @@ def _hermetic(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Generator[None patch("gatelock._randr.RandrBackend.create", return_value=None), patch("gatelock._outputs.scan_xrandr", return_value=DEFAULT_OUTPUTS), patch("gatelock._detect._RandrEventSource.start", return_value=False), + # A lock that yields to a prompt asks X what is on screen; no test may + # reach the developer's display for that. + patch("gatelock._prompts_x.load_xlib", return_value=None), + patch("gatelock._grabprobe.load_xlib", return_value=None), + # The startup grace is real time; the queue tests pass it explicitly. + patch("gatelock._queue.STARTUP_GRACE_SECONDS", 0.0), patch("gatelock._vt.shutil.which", return_value="/usr/bin/setxkbmap"), patch("gatelock._vt.subprocess.run"), ): @@ -162,3 +169,39 @@ def build_loop( def loop(mock_root: MagicMock) -> tuple[RecoveryLoop, MagicMock, SurfaceSet]: """A recovery loop in hard mode, the configuration most tests want.""" return build_loop(mock_root, LockConfig(mode="hard")) + + +_MADE: list[Arbiter] = [] + + +def make_arbiter( + runtime: Path, app: str, rank: int, *, grab: str = "global", vt: bool = True +) -> Arbiter: + """A published arbiter in ``runtime``; released after the test.""" + arbiter = Arbiter(app, rank, grab=grab, disable_vt=vt, runtime_dir=runtime) + arbiter.publish() + _MADE.append(arbiter) + return arbiter + + +@pytest.fixture(autouse=True) +def _release_arbiters() -> Generator[None]: + """Close every claim file a test opened.""" + yield + while _MADE: + _MADE.pop().release() + + +@pytest.fixture +def armed( + mock_root: MagicMock, tmp_path: Path +) -> Generator[tuple[LockWindow, Arbiter]]: + """A mid-rank (150) hard lock that is armed, holds the screen, and checks.""" + arbiter = make_arbiter(tmp_path, "leetcode_guard", 150) + arbiter.acquire_holder() + config = LockConfig(mode="hard", app_name="leetcode_guard", rank=150) + window = LockWindow(mock_root, config, MagicMock(), arbiter=arbiter) + window.setup() + window.grab_input() + yield window, arbiter + window.close() diff --git a/gatelock/gatelock/tests/test_grab_blame.py b/gatelock/gatelock/tests/test_grab_blame.py new file mode 100644 index 0000000..5ad7aa7 --- /dev/null +++ b/gatelock/gatelock/tests/test_grab_blame.py @@ -0,0 +1,202 @@ +# Copyright (c) 2026 Krzysztof Rudnicki +"""Tests for WHO a blocked grab is blamed on, and the error text that explains it. + +Incident 2026-11-11: the lock logged "held by gatelock app 'leetcode_guard' +(pid )" -- it blamed itself, because the holder file it had just +written is the one ``describe_holder`` reads -- while the real holder was the +GNOME Keyring prompt, an ordinary X client. Tk's own error text ("another +application has grab" vs "window not viewable") was swallowed as well. +""" + +from __future__ import annotations + +from dataclasses import replace +import logging +import os +import tkinter as tk +from typing import TYPE_CHECKING +from unittest.mock import MagicMock, patch + +from gatelock import _preempt +from gatelock._arbiter import RANK_SCREEN_LOCKER, Arbiter +from gatelock._config import GrabPolicy, LockConfig +from gatelock._prompts import PromptYield +from gatelock.tests.conftest import make_window + +if TYPE_CHECKING: + from pathlib import Path + + import pytest + +REASON = "grab failed: another application has grab" + + +def _own_arbiter(tmp_path: Path) -> Arbiter: + """A real arbiter that holds the screen, as a running lock does.""" + arbiter = Arbiter( + "leetcode_guard", + RANK_SCREEN_LOCKER, + grab="global", + disable_vt=True, + runtime_dir=tmp_path, + ) + arbiter.publish() + assert arbiter.acquire_holder() + return arbiter + + +class TestNoSelfBlame: + """When the holder file is our own, the grab holder is somebody else.""" + + def test_own_claim_is_not_reported_as_the_holder( + self, tmp_path: Path, caplog: pytest.LogCaptureFixture + ) -> None: + """No "held by gatelock app", no own pid, and a pointer at X clients.""" + arbiter = _own_arbiter(tmp_path) + config = LockConfig(mode="hard", grab=GrabPolicy(preempt_weaker_holder=True)) + with ( + caplog.at_level(logging.WARNING), + patch("gatelock._preempt.os.kill") as kill, + ): + _preempt.log_grab_blocked( + 25, arbiter=arbiter, config=config, preempted_pids=set() + ) + arbiter.release() + assert "held by gatelock app" not in caplog.text + assert f"pid {os.getpid()}" not in caplog.text + assert "non-gatelock X client" in caplog.text + kill.assert_not_called() + + def test_prompt_on_screen_is_named( + self, tmp_path: Path, caplog: pytest.LogCaptureFixture + ) -> None: + """A detected credential prompt is named as present, not as a verdict.""" + arbiter = _own_arbiter(tmp_path) + with caplog.at_level(logging.WARNING): + _preempt.log_grab_blocked( + 25, + arbiter=arbiter, + config=LockConfig(), + preempted_pids=set(), + prompt_yield=MagicMock(last_reason="", prompt_names=("gcr-prompter",)), + ) + arbiter.release() + assert "gcr-prompter" in caplog.text + + def test_without_arbiter_names_the_prompt_too( + self, caplog: pytest.LogCaptureFixture + ) -> None: + """Arbitration disabled: same message, same prompt.""" + with caplog.at_level(logging.WARNING): + _preempt.log_grab_blocked( + 25, + arbiter=None, + config=LockConfig(), + preempted_pids=set(), + prompt_yield=MagicMock(last_reason="", prompt_names=("pinentry",)), + ) + assert "pinentry" in caplog.text + assert "fullscreen game" in caplog.text + + def test_another_gatelock_app_is_still_named( + self, caplog: pytest.LogCaptureFixture + ) -> None: + """Only the self-reference changed: a real foreign holder is still blamed.""" + arbiter = MagicMock() + arbiter.claim.instance_id = "ours" + arbiter.describe_holder.return_value = MagicMock( + app="diet_guard", rank=100, pid=4242, started="t", instance_id="theirs" + ) + with caplog.at_level(logging.WARNING): + _preempt.log_grab_blocked( + 25, arbiter=arbiter, config=LockConfig(), preempted_pids=set() + ) + assert "held by gatelock app 'diet_guard'" in caplog.text + + +class TestReasonInLogs: + """Tk's own message rides along on every blocked-grab warning.""" + + def test_reason_with_no_gatelock_holder( + self, caplog: pytest.LogCaptureFixture + ) -> None: + """The no-holder branch carries it.""" + with caplog.at_level(logging.WARNING): + _preempt.log_grab_blocked( + 25, + arbiter=None, + config=LockConfig(), + preempted_pids=set(), + prompt_yield=MagicMock(last_reason=REASON, prompt_names=()), + ) + assert REASON in caplog.text + + def test_reason_with_a_gatelock_holder( + self, caplog: pytest.LogCaptureFixture + ) -> None: + """The holder branch carries it too.""" + arbiter = MagicMock() + arbiter.claim.instance_id = "ours" + arbiter.describe_holder.return_value = MagicMock( + app="diet_guard", rank=100, pid=4242, started="t", instance_id="theirs" + ) + with caplog.at_level(logging.WARNING): + _preempt.log_grab_blocked( + 25, + arbiter=arbiter, + config=LockConfig(), + preempted_pids=set(), + prompt_yield=MagicMock(last_reason=REASON, prompt_names=()), + ) + assert REASON in caplog.text + + def test_window_passes_the_last_reason_and_prompts( + self, mock_root: MagicMock, caplog: pytest.LogCaptureFixture + ) -> None: + """``LockWindow`` hands the policy's state to the logger.""" + window, _ = make_window(mock_root, config=LockConfig(mode="hard")) + stub = MagicMock(last_reason=REASON, prompt_names=("gcr-prompter",)) + window._arming = replace(window._arming, prompt_yield=stub) + with caplog.at_level(logging.WARNING): + window._log_grab_blocked(25) + assert REASON in caplog.text + assert "gcr-prompter" in caplog.text + + +class TestArming: + """The arming retry loop feeds the same policy and surfaces the error.""" + + def test_failures_feed_the_policy(self, mock_root: MagicMock) -> None: + """Each failed attempt records Tk's message.""" + mock_root.grab_set_global.side_effect = tk.TclError(REASON) + window, _ = make_window(mock_root, config=LockConfig(mode="hard")) + window._acquire_global_grab(attempt=1) + assert window._arming.prompt_yield.last_reason == REASON + + def test_local_fallback_warning_has_the_reason( + self, mock_root: MagicMock, caplog: pytest.LogCaptureFixture + ) -> None: + """``retry_ms=0``: the fallback says why the global grab failed.""" + mock_root.grab_set_global.side_effect = tk.TclError(REASON) + config = LockConfig(mode="hard", grab=GrabPolicy(retry_ms=0)) + window, _ = make_window(mock_root, config=config) + with caplog.at_level(logging.WARNING): + window._acquire_global_grab(attempt=1) + assert REASON in caplog.text + + def test_success_ends_a_yield_and_relifts_the_lock( + self, mock_root: MagicMock + ) -> None: + """The prompt closed and the retry won: the lock covers the screen again.""" + window, _ = make_window(mock_root, config=LockConfig(mode="hard")) + policy = MagicMock(spec=PromptYield) + policy.grab_held.return_value = True + window._arming = replace(window._arming, prompt_yield=policy) + window._acquire_global_grab(attempt=3) + policy.grab_held.assert_called_once_with() + mock_root.lift.assert_called() + + def test_window_shares_one_policy(self, mock_root: MagicMock) -> None: + """Arming and recovery must time the same block.""" + window, _ = make_window(mock_root, config=LockConfig(mode="hard")) + assert window.recovery._parts.prompt_yield is window._arming.prompt_yield diff --git a/gatelock/gatelock/tests/test_grabprobe.py b/gatelock/gatelock/tests/test_grabprobe.py new file mode 100644 index 0000000..7d15fdb --- /dev/null +++ b/gatelock/gatelock/tests/test_grabprobe.py @@ -0,0 +1,238 @@ +# Copyright (c) 2026 Krzysztof Rudnicki +"""Tests for the non-blocking grab probe (:mod:`gatelock._grabprobe`). + +Tk's ``grab -global`` blocks the whole event loop for ~1 s per FAILED attempt. +The probe asks X first, on a throwaway connection, so a lock whose grab is held +by someone else skips the Tk call instead of freezing its own UI. python-xlib +is replaced by an in-memory fake; ``scripts/verify_grab_responsive.py`` runs the +real thing against a real foreign grab. +""" + +from __future__ import annotations + +from dataclasses import replace +import logging +import tkinter as tk +from types import SimpleNamespace +from typing import TYPE_CHECKING +from unittest.mock import MagicMock, patch + +import pytest + +from gatelock._config import LockConfig +from gatelock._grabprobe import GrabProbe +from gatelock._prompts import PromptYield +from gatelock.tests.conftest import BOTH, build_loop, make_window + +if TYPE_CHECKING: + from collections.abc import Callable + +OK, TAKEN, NOT_VIEWABLE = 0, 1, 3 + + +class XError(Exception): + """Stands in for ``Xlib.error.XError``.""" + + +class DisplayError(Exception): + """Stands in for ``Xlib.error.DisplayError``.""" + + +class FakeRoot: + """Answers grab requests from a script and records the ungrabs.""" + + def __init__(self, keyboard: int, pointer: int) -> None: + self.statuses = {"keyboard": keyboard, "pointer": pointer} + self.asked: list[str] = [] + + def grab_keyboard(self, *_args: object) -> int: + self.asked.append("keyboard") + return self.statuses["keyboard"] + + def grab_pointer(self, *_args: object) -> int: + self.asked.append("pointer") + if self.statuses["pointer"] < 0: + message = "BadWindow" + raise XError(message) + return self.statuses["pointer"] + + +class FakeDisplay: + """The slice of ``Xlib.display.Display`` the probe uses.""" + + def __init__(self, root: FakeRoot) -> None: + self.root = root + self.ungrabbed: list[str] = [] + self.closed = False + + def screen(self) -> SimpleNamespace: + return SimpleNamespace(root=self.root) + + def ungrab_keyboard(self, _time: int) -> None: + self.ungrabbed.append("keyboard") + + def ungrab_pointer(self, _time: int) -> None: + self.ungrabbed.append("pointer") + + def sync(self) -> None: + self.ungrabbed.append("sync") + + def close(self) -> None: + self.closed = True + + +def loaded(factory: Callable[[], FakeDisplay]) -> tuple[SimpleNamespace, ...]: + """What ``load_xlib("Xlib.display", "Xlib.X", "Xlib.error")`` returns.""" + return ( + SimpleNamespace(Display=factory), + SimpleNamespace( + GrabSuccess=OK, + AlreadyGrabbed=TAKEN, + GrabModeAsync=1, + CurrentTime=0, + NONE=0, + ), + SimpleNamespace( + DisplayError=DisplayError, + XauthError=DisplayError, + ConnectionClosedError=DisplayError, + XError=XError, + ), + ) + + +def probe_with(display: FakeDisplay) -> tuple[GrabProbe, object]: + """A probe whose X connection is ``display`` (patch is the second item).""" + patcher = patch( + "gatelock._grabprobe.load_xlib", return_value=loaded(lambda: display) + ) + patcher.start() + return GrabProbe(), patcher + + +@pytest.fixture(autouse=True) +def _unpatch() -> object: + yield + patch.stopall() + + +class TestProbe: + """What the probe reports, and that it leaves nothing grabbed.""" + + def test_free_grab_is_released_at_once(self) -> None: + """Nobody holds it: report free, and give the probe's own grab back.""" + display = FakeDisplay(FakeRoot(OK, OK)) + probe, _ = probe_with(display) + assert probe.blocked() is None + assert display.root.asked == ["keyboard", "pointer"] + assert display.ungrabbed == ["keyboard", "pointer", "sync"] + assert display.closed + + @pytest.mark.parametrize( + ("keyboard", "pointer", "which"), + [(TAKEN, OK, "keyboard"), (OK, TAKEN, "pointer")], + ) + def test_foreign_grab_is_reported( + self, keyboard: int, pointer: int, which: str + ) -> None: + """Either device held elsewhere blocks us; the reason names it.""" + display = FakeDisplay(FakeRoot(keyboard, pointer)) + probe, _ = probe_with(display) + reason = probe.blocked() + assert reason is not None + assert "another application has grab" in reason + assert which in reason + assert display.ungrabbed[-1] == "sync" + assert display.root.asked[-1] == which + + def test_other_statuses_defer_to_tk(self) -> None: + """Not viewable and the like are not a foreign holder: let Tk decide.""" + display = FakeDisplay(FakeRoot(NOT_VIEWABLE, OK)) + probe, _ = probe_with(display) + assert probe.blocked() is None + + def test_check_raises_tcl_error_when_blocked(self) -> None: + """``check`` speaks Tk's dialect so callers keep one except clause.""" + probe, _ = probe_with(FakeDisplay(FakeRoot(TAKEN, OK))) + with pytest.raises(tk.TclError, match="another application has grab"): + probe.check() + + def test_check_is_silent_when_free(self) -> None: + """Free grab: nothing raised.""" + probe, _ = probe_with(FakeDisplay(FakeRoot(OK, OK))) + probe.check() + + +class TestFailClosed: + """When X cannot be asked, fall back to today's behaviour: call Tk.""" + + def test_missing_xlib(self, caplog: pytest.LogCaptureFixture) -> None: + """No python-xlib: unknown, said once.""" + probe = GrabProbe() + with ( + patch("gatelock._grabprobe.load_xlib", return_value=None), + caplog.at_level(logging.WARNING), + ): + assert probe.blocked() is None + assert probe.blocked() is None + assert caplog.text.count("python-xlib") == 1 + + def test_no_connection(self, caplog: pytest.LogCaptureFixture) -> None: + """A connection that will not open: unknown, warned.""" + + def refuse() -> FakeDisplay: + message = "no display" + raise DisplayError(message) + + probe = GrabProbe() + with ( + patch("gatelock._grabprobe.load_xlib", return_value=loaded(refuse)), + caplog.at_level(logging.WARNING), + ): + assert probe.blocked() is None + assert "X connection" in caplog.text + + def test_request_error_still_closes(self, caplog: pytest.LogCaptureFixture) -> None: + """An X error mid-probe: unknown, warned, connection dropped.""" + display = FakeDisplay(FakeRoot(OK, -1)) + probe, _ = probe_with(display) + with caplog.at_level(logging.WARNING): + assert probe.blocked() is None + assert "BadWindow" in caplog.text + assert display.closed + + +class TestWiring: + """The probe gates every global-grab attempt, in arming and in recovery.""" + + def test_policy_exposes_check_grab(self) -> None: + """``PromptYield.check_grab`` is the one entry point.""" + probe = MagicMock() + PromptYield(grab_probe=probe).check_grab() + probe.check.assert_called_once_with() + + def test_arming_skips_tk_when_blocked(self, mock_root: MagicMock) -> None: + """A foreign holder: the 1 s Tk call is never made, the reason is kept.""" + window, _ = make_window(mock_root, config=LockConfig(mode="hard")) + policy = PromptYield(grab_probe=MagicMock()) + policy._grab_probe.check.side_effect = tk.TclError("held") + window._arming = replace(window._arming, prompt_yield=policy) + window._acquire_global_grab(attempt=1) + mock_root.grab_set_global.assert_not_called() + assert policy.last_reason == "held" + mock_root.after.assert_called() + + def test_recovery_skips_tk_when_blocked(self, mock_root: MagicMock) -> None: + """Same in the recovery tick.""" + recovery, enumerator, _ = build_loop(mock_root, LockConfig(mode="hard")) + enumerator.scan.return_value = BOTH + mock_root.grab_current.return_value = None + policy = PromptYield(grab_probe=MagicMock()) + policy._grab_probe.check.side_effect = tk.TclError("held") + recovery._parts = recovery._parts.__class__( + **{**vars(recovery._parts), "prompt_yield": policy} + ) + report = recovery.tick() + mock_root.grab_set_global.assert_not_called() + assert report.grab_reasserted is False + assert policy.last_reason == "held" diff --git a/gatelock/gatelock/tests/test_prompts.py b/gatelock/gatelock/tests/test_prompts.py new file mode 100644 index 0000000..7691d70 --- /dev/null +++ b/gatelock/gatelock/tests/test_prompts.py @@ -0,0 +1,214 @@ +# Copyright (c) 2026 Krzysztof Rudnicki +"""Tests for the yield-to-system-prompt policy (:mod:`gatelock._prompts`). + +Incident 2026-11-11: the GNOME Keyring prompt held the X grab, the lock +covered it, and nothing was clickable. The policy: once the grab has been +blocked for a couple of seconds AND an allowlisted credential prompt is on +screen, raise that prompt above the lock and stop covering it again. +""" + +from __future__ import annotations + +import logging + +import pytest + +from gatelock._prompt_names import ( + SYSTEM_PROMPT_CLASSES, + PromptWindow, + is_system_prompt, +) +from gatelock._prompts import YIELD_AFTER_S, PromptYield +from gatelock._prompts_x import XPromptProbe + +GCR = PromptWindow( + window_id=11, toplevel_id=10, wm_class=("gcr-prompter", "Gcr-prompter") +) + + +class FakeClock: + """A settable monotonic clock.""" + + def __init__(self) -> None: + self.now = 100.0 + + def __call__(self) -> float: + return self.now + + +class FakeProbe: + """Answers ``raise_prompts`` from a script and counts the calls.""" + + def __init__(self, *prompts: PromptWindow) -> None: + self.prompts = prompts + self.calls = 0 + + def raise_prompts(self) -> tuple[PromptWindow, ...]: + self.calls += 1 + return self.prompts + + +class TestAllowlist: + """Only known credential prompts are ever yielded to.""" + + @pytest.mark.parametrize( + "wm_class", + [ + ("gcr-prompter", "Gcr-prompter"), + ("pinentry-gtk-2", "Pinentry-gtk-2"), + ("pinentry", "Pinentry"), + ( + "polkit-gnome-authentication-agent-1", + "Polkit-gnome-authentication-agent-1", + ), + ("lxpolkit", "Lxpolkit"), + ("ssh-askpass", "SSH-Askpass"), + ("x11-ssh-askpass", "X11-ssh-askpass"), + ("ksshaskpass", "ksshaskpass"), + ("gnome-ssh-askpass", "Gnome-ssh-askpass"), + ("whatever", "GCR-PROMPTER"), + ("Pinentry-Qt", "x"), + ], + ) + def test_known_prompt_matches(self, wm_class: tuple[str, str]) -> None: + """Instance or class, any case, with the documented globs.""" + assert is_system_prompt(wm_class) + + @pytest.mark.parametrize( + "wm_class", + [ + ("steam_game", "Steam_game"), + ("firefox", "Firefox"), + ("tk", "Tk"), + ("pinentr", "x"), + ("xpinentry", "x"), + ("", ""), + ], + ) + def test_anything_else_does_not(self, wm_class: tuple[str, str]) -> None: + """A fullscreen game must keep today's behaviour: lock stays on top.""" + assert not is_system_prompt(wm_class) + + def test_the_allowlist_is_one_named_constant(self) -> None: + """Every pattern the policy approved is in ``SYSTEM_PROMPT_CLASSES``.""" + assert set(SYSTEM_PROMPT_CLASSES) == { + "gcr-prompter", + "pinentry*", + "polkit-gnome-authentication-agent-1", + "lxpolkit", + "ssh-askpass*", + "x11-ssh-askpass", + "ksshaskpass", + "gnome-ssh-askpass", + } + + def test_prompt_window_name(self) -> None: + """The log names the window by its WM_CLASS instance.""" + assert GCR.name == "gcr-prompter" + + +class TestYieldPolicy: + """When the lock steps aside, and when it does not.""" + + def _yielder( + self, *prompts: PromptWindow + ) -> tuple[PromptYield, FakeClock, FakeProbe]: + clock, probe = FakeClock(), FakeProbe(*prompts) + return PromptYield(probe, clock=clock), clock, probe + + def test_default_probe_is_the_xlib_one(self) -> None: + """Without an explicit probe the real X one is used.""" + assert isinstance(PromptYield()._probe, XPromptProbe) + + def test_idle_before_anything_is_blocked(self) -> None: + """No failed grab, no look at the screen.""" + policy, _, probe = self._yielder(GCR) + policy.settle() + assert not policy.yielding + assert probe.calls == 0 + + def test_waits_out_the_delay(self) -> None: + """A grab blocked for less than the delay is a blip, not a prompt.""" + policy, clock, probe = self._yielder(GCR) + policy.grab_blocked("another application has grab") + clock.now += YIELD_AFTER_S - 0.1 + policy.settle() + assert not policy.yielding + assert probe.calls == 0 + + def test_yields_to_a_prompt_after_the_delay( + self, caplog: pytest.LogCaptureFixture + ) -> None: + """Blocked past the delay with a prompt on screen: yield, and say so.""" + policy, clock, _ = self._yielder(GCR) + policy.grab_blocked("another application has grab") + clock.now += YIELD_AFTER_S + 0.1 + with caplog.at_level(logging.WARNING): + policy.settle() + assert policy.yielding + assert policy.prompt_names == ("gcr-prompter",) + assert "gcr-prompter" in caplog.text + + def test_keeps_raising_it_every_settle(self) -> None: + """Each tick re-raises the prompt, last, over anything the tick lifted.""" + policy, clock, probe = self._yielder(GCR) + policy.grab_blocked("x") + clock.now += 5 + policy.settle() + policy.settle() + assert probe.calls == 2 + assert policy.yielding + + def test_no_prompt_means_no_yield(self) -> None: + """A game holding the grab: the probe finds nothing, the lock stays on top.""" + policy, clock, probe = self._yielder() + policy.grab_blocked("another application has grab") + clock.now += 30 + policy.settle() + assert not policy.yielding + assert policy.prompt_names == () + assert probe.calls == 1 + + def test_stops_when_the_prompt_goes_away( + self, caplog: pytest.LogCaptureFixture + ) -> None: + """Prompt unmapped but grab still blocked: stack the lock normally again.""" + policy, clock, probe = self._yielder(GCR) + policy.grab_blocked("x") + clock.now += 5 + policy.settle() + probe.prompts = () + with caplog.at_level(logging.INFO): + policy.settle() + assert not policy.yielding + assert "gone" in caplog.text + + def test_the_clock_starts_at_the_first_failure(self) -> None: + """Later failures must not restart the delay.""" + policy, clock, _ = self._yielder(GCR) + policy.grab_blocked("first") + clock.now += 1.5 + policy.grab_blocked("second") + clock.now += 0.6 + policy.settle() + assert policy.yielding + assert policy.last_reason == "second" + + def test_grab_held_ends_the_yield_and_says_so(self) -> None: + """Taking the grab reports whether the lock must now re-raise itself.""" + policy, clock, probe = self._yielder(GCR) + policy.grab_blocked("x") + clock.now += 5 + policy.settle() + assert policy.grab_held() is True + assert not policy.yielding + assert policy.prompt_names == () + policy.settle() + assert probe.calls == 1 # nothing blocked any more: no more X work + + def test_grab_held_when_never_yielding(self) -> None: + """The common case: nothing to undo.""" + policy, _, _ = self._yielder(GCR) + assert policy.grab_held() is False + policy.grab_blocked("x") + assert policy.grab_held() is False diff --git a/gatelock/gatelock/tests/test_prompts_x.py b/gatelock/gatelock/tests/test_prompts_x.py new file mode 100644 index 0000000..eb11ce4 --- /dev/null +++ b/gatelock/gatelock/tests/test_prompts_x.py @@ -0,0 +1,209 @@ +# Copyright (c) 2026 Krzysztof Rudnicki +"""Tests for finding and raising a credential prompt over python-xlib. + +python-xlib is optional and there is no X server in the suite, so the library +is replaced by a small in-memory window tree. The real thing is exercised by +``scripts/verify_prompt_yield.py`` on a private Xvfb. +""" + +from __future__ import annotations + +from contextlib import contextmanager +from types import SimpleNamespace +from typing import TYPE_CHECKING +from unittest.mock import patch + +from gatelock._prompts_x import XPromptProbe + +if TYPE_CHECKING: + from collections.abc import Callable, Iterator + + from gatelock._prompt_names import PromptWindow + +ABOVE = 0 +ANY = 0 +VIEWABLE = 2 + + +class XError(Exception): + """Stands in for ``Xlib.error.XError``.""" + + +class DisplayError(Exception): + """Stands in for ``Xlib.error.DisplayError``.""" + + +class ClosedError(Exception): + """Stands in for ``Xlib.error.ConnectionClosedError``.""" + + +class NoAuthError(Exception): + """Stands in for ``Xlib.error.XauthError``.""" + + +class FakeWindow: + """One node of the fake window tree.""" + + def __init__( + self, + display: FakeDisplay, + wid: int, + wm_class: tuple[str, str] | None = None, + *, + viewable: bool = True, + ) -> None: + self.display = display + self.id = wid + self.wm_class = wm_class + self.viewable = viewable + self.parent: FakeWindow | None = None + self.children: list[FakeWindow] = [] + self.client_list: list[int] | None = None + self.raised = False + self.broken = False + self.configure_error: XError | None = None + display.windows[wid] = self + + def add(self, child: FakeWindow) -> FakeWindow: + child.parent = self + self.children.append(child) + return child + + def get_wm_class(self) -> tuple[str, str] | None: + if self.broken: + message = "BadWindow" + raise XError(message) + return self.wm_class + + def get_attributes(self) -> SimpleNamespace: + return SimpleNamespace(map_state=VIEWABLE if self.viewable else 0) + + def query_tree(self) -> SimpleNamespace: + return SimpleNamespace(children=list(self.children), parent=self.parent) + + def get_full_property(self, atom: str, _kind: int) -> SimpleNamespace | None: + if atom != "_NET_CLIENT_LIST" or self.client_list is None: + return None + return SimpleNamespace(value=self.client_list) + + def configure(self, *, stack_mode: int, onerror: Callable[[XError], None]) -> None: + assert stack_mode == ABOVE + self.raised = True + if self.configure_error is not None: + onerror(self.configure_error) + + +class FakeDisplay: + """The slice of ``Xlib.display.Display`` the probe uses.""" + + def __init__(self) -> None: + self.windows: dict[int, FakeWindow] = {} + self.root = FakeWindow(self, 1) + self.closed = False + self.synced = False + + def screen(self) -> SimpleNamespace: + return SimpleNamespace(root=self.root) + + def intern_atom(self, name: str) -> str: + return name + + def create_resource_object(self, _kind: str, wid: int) -> FakeWindow: + return self.windows[wid] + + def sync(self) -> None: + self.synced = True + + def close(self) -> None: + self.closed = True + + +def modules(display_factory: Callable[[], FakeDisplay]) -> tuple[SimpleNamespace, ...]: + """What ``load_xlib("Xlib.display", "Xlib.X", "Xlib.error")`` returns.""" + return ( + SimpleNamespace(Display=display_factory), + SimpleNamespace(Above=ABOVE, AnyPropertyType=ANY, IsViewable=VIEWABLE), + SimpleNamespace( + DisplayError=DisplayError, + XauthError=NoAuthError, + ConnectionClosedError=ClosedError, + XError=XError, + ), + ) + + +@contextmanager +def probe_over(display: FakeDisplay) -> Iterator[XPromptProbe]: + """A probe whose X connection is ``display``.""" + loaded = modules(lambda: display) + with patch("gatelock._prompts_x.load_xlib", return_value=loaded): + yield XPromptProbe() + + +def scan(display: FakeDisplay) -> tuple[PromptWindow, ...]: + """One ``raise_prompts()`` over ``display``.""" + with probe_over(display) as probe: + return probe.raise_prompts() + + +class TestFinding: + """Which windows count, and what gets raised.""" + + def test_framed_client_raises_the_frame(self) -> None: + """i3 reparents: the class is on the client, the frame is the top-level.""" + display = FakeDisplay() + frame = display.root.add(FakeWindow(display, 10, ("i3-frame", "i3-frame"))) + client = frame.add(FakeWindow(display, 11, ("gcr-prompter", "Gcr-prompter"))) + found = scan(display) + assert [(p.window_id, p.toplevel_id) for p in found] == [(11, 10)] + assert frame.raised + assert not client.raised + assert display.synced + assert display.closed + + def test_unframed_client_is_its_own_toplevel(self) -> None: + """Without a window manager the prompt is a direct root child.""" + display = FakeDisplay() + prompt = display.root.add( + FakeWindow(display, 20, ("pinentry-gtk-2", "Pinentry")) + ) + found = scan(display) + assert [(p.window_id, p.toplevel_id) for p in found] == [(20, 20)] + assert prompt.raised + + def test_net_client_list_is_used_and_deduplicated(self) -> None: + """A client listed by the WM and also found in the tree is raised once.""" + display = FakeDisplay() + frame = display.root.add(FakeWindow(display, 10)) + frame.add(FakeWindow(display, 11, ("gcr-prompter", "x"))) + display.root.client_list = [11, 11] + found = scan(display) + assert [p.window_id for p in found] == [11] + + def test_net_client_list_alone_finds_a_deeper_window(self) -> None: + """A client the tree walk would not reach is still found via the WM list.""" + display = FakeDisplay() + outer = display.root.add(FakeWindow(display, 10)) + mid = outer.add(FakeWindow(display, 11)) + deep = mid.add(FakeWindow(display, 12, ("lxpolkit", "Lxpolkit"))) + display.root.client_list = [12] + found = scan(display) + assert [(p.window_id, p.toplevel_id) for p in found] == [(12, 10)] + assert outer.raised + assert not deep.raised + + def test_unmapped_prompt_is_ignored(self) -> None: + """A prompt on another workspace is not on screen: leave the lock alone.""" + display = FakeDisplay() + display.root.add(FakeWindow(display, 10, ("gcr-prompter", "x"), viewable=False)) + assert scan(display) == () + + def test_other_windows_are_ignored(self) -> None: + """No class, or a class off the allowlist, is not a prompt.""" + display = FakeDisplay() + display.root.add(FakeWindow(display, 10, ("steam_game", "Steam"))) + display.root.add(FakeWindow(display, 11, None)) + found = scan(display) + assert found == () + assert not display.windows[10].raised + assert display.closed diff --git a/gatelock/gatelock/tests/test_prompts_x_failures.py b/gatelock/gatelock/tests/test_prompts_x_failures.py new file mode 100644 index 0000000..07e01ab --- /dev/null +++ b/gatelock/gatelock/tests/test_prompts_x_failures.py @@ -0,0 +1,98 @@ +# Copyright (c) 2026 Krzysztof Rudnicki +"""Failure handling of :class:`XPromptProbe`: every error fails closed. + +Split from ``test_prompts_x.py`` (250-line cap), whose in-memory window tree +and python-xlib stand-ins it reuses. +""" + +from __future__ import annotations + +import logging +from unittest.mock import patch + +import pytest + +from gatelock._prompts_x import XPromptProbe +from gatelock.tests.test_prompts_x import ( + ClosedError, + DisplayError, + FakeDisplay, + FakeWindow, + NoAuthError, + XError, + modules, + probe_over, +) + + +class TestFailures: + """Anything going wrong fails closed: no prompt found, lock stays on top.""" + + def test_missing_xlib_is_reported_once( + self, caplog: pytest.LogCaptureFixture + ) -> None: + """No python-xlib: nothing to do, said once.""" + probe = XPromptProbe() + with ( + patch("gatelock._prompts_x.load_xlib", return_value=None), + caplog.at_level(logging.WARNING), + ): + assert probe.raise_prompts() == () + assert probe.raise_prompts() == () + assert caplog.text.count("python-xlib") == 1 + + @pytest.mark.parametrize( + "error", + [OSError("gone"), DisplayError("no display"), ClosedError(), NoAuthError()], + ) + def test_no_connection( + self, error: Exception, caplog: pytest.LogCaptureFixture + ) -> None: + """An X connection that will not open is a warning, not a crash.""" + + def refuse() -> FakeDisplay: + raise error + + probe = XPromptProbe() + with ( + patch("gatelock._prompts_x.load_xlib", return_value=modules(refuse)), + caplog.at_level(logging.WARNING), + ): + assert probe.raise_prompts() == () + assert "X connection" in caplog.text + + def test_window_vanishing_mid_scan(self, caplog: pytest.LogCaptureFixture) -> None: + """BadWindow from a window that went away aborts this scan, once-warned.""" + display = FakeDisplay() + display.root.add(FakeWindow(display, 10, ("gcr-prompter", "x"))).broken = True + with probe_over(display) as probe, caplog.at_level(logging.WARNING): + assert probe.raise_prompts() == () + assert probe.raise_prompts() == () + assert caplog.text.count("BadWindow") == 1 + assert display.closed + + def test_a_good_scan_rearms_the_warning( + self, caplog: pytest.LogCaptureFixture + ) -> None: + """After a success the same failure is news again.""" + display = FakeDisplay() + window = display.root.add(FakeWindow(display, 10, ("gcr-prompter", "x"))) + with probe_over(display) as probe, caplog.at_level(logging.WARNING): + window.broken = True + probe.raise_prompts() + window.broken = False + assert probe.raise_prompts() + window.broken = True + probe.raise_prompts() + assert caplog.text.count("BadWindow") == 2 + + def test_raise_refused_is_not_a_yield( + self, caplog: pytest.LogCaptureFixture + ) -> None: + """If the X server rejects the restack, claim nothing.""" + display = FakeDisplay() + prompt = display.root.add(FakeWindow(display, 10, ("gcr-prompter", "x"))) + prompt.configure_error = XError("BadMatch") + with probe_over(display) as probe, caplog.at_level(logging.WARNING): + assert probe.raise_prompts() == () + assert "BadMatch" in caplog.text diff --git a/gatelock/gatelock/tests/test_recovery_invariants.py b/gatelock/gatelock/tests/test_recovery_invariants.py index 192ed50..ca8f3b3 100644 --- a/gatelock/gatelock/tests/test_recovery_invariants.py +++ b/gatelock/gatelock/tests/test_recovery_invariants.py @@ -64,3 +64,17 @@ class TestStaticInvariant: assert "self._sink.put" in loop_body for forbidden in ("geometry(", "deiconify(", "lift(", "grab_"): assert forbidden not in loop_body + + def test_stepback_is_the_only_release_path(self) -> None: + """Only the step-back module may release the grab. + + It triggers on a strictly stronger, live, at-least-as-firm claim and + nowhere else; any other module growing a release call is a bypass. + """ + callers = { + path.name + for path in SOURCE_ROOT.glob("*.py") + for node in ast.walk(ast.parse(path.read_text(encoding="utf-8"))) + if isinstance(node, ast.Attribute) and node.attr == "grab_release" + } + assert callers == {"_stepback.py"} diff --git a/gatelock/gatelock/tests/test_recovery_restart.py b/gatelock/gatelock/tests/test_recovery_restart.py new file mode 100644 index 0000000..5c6e86d --- /dev/null +++ b/gatelock/gatelock/tests/test_recovery_restart.py @@ -0,0 +1,43 @@ +# Copyright (c) 2026 Krzysztof Rudnicki +"""An output that comes up while the recovery loop is stopped is covered on restart. + +leetcode-guard's study mode stops the recovery loop and the change detector on +purpose. On 2026-11-11 the 4K output (DP-2) only got a mode 3m20s after the +lock armed -- long after study mode had stopped both -- so the backdrop stayed +6400x1440 and no surface was built for it. That is the stopped loop doing what +it was told, not a detection bug: this pins that the FIRST tick after +``start()`` sizes the backdrop to the new screen and builds the missing surface. +""" + +from __future__ import annotations + +from typing import TYPE_CHECKING + +from gatelock._config import LockConfig +from gatelock._outputs import OutputScan +from gatelock.tests.conftest import BOTH, HDMI, build_loop + +if TYPE_CHECKING: + from unittest.mock import MagicMock + +HDMI_ONLY = OutputScan((HDMI,), "randr", ok=True) + + +def test_output_that_appeared_while_stopped_is_covered_on_restart( + mock_root: MagicMock, +) -> None: + """Stop, grow the screen, start: one tick later the new output is locked.""" + recovery, enumerator, surfaces = build_loop(mock_root, LockConfig(mode="hard")) + enumerator.scan.return_value = HDMI_ONLY + mock_root.winfo_screenheight.return_value = 1440 + recovery.tick() + assert surfaces.names() == {"HDMI-0"} + + recovery.stop() + enumerator.scan.return_value = BOTH # DP-0 is the 4K output here + mock_root.winfo_screenwidth.return_value = 6400 + mock_root.winfo_screenheight.return_value = 2160 + recovery.tick() # study mode off: the first tick after start() is this one + + assert surfaces.names() == {"DP-0", "HDMI-0"} + mock_root.geometry.assert_called_with("6400x2160+0+0") diff --git a/gatelock/gatelock/tests/test_recovery_yield.py b/gatelock/gatelock/tests/test_recovery_yield.py new file mode 100644 index 0000000..a36bae7 --- /dev/null +++ b/gatelock/gatelock/tests/test_recovery_yield.py @@ -0,0 +1,204 @@ +# Copyright (c) 2026 Krzysztof Rudnicki +"""Tests for how a recovery tick treats a blocked grab and a waiting prompt. + +The tick must (a) say WHY the re-grab failed, (b) feed the yield policy, +(c) stop lifting the lock over a prompt it is yielding to, and (d) re-raise +the prompt last, so nothing the tick lifted covers it. +""" + +from __future__ import annotations + +from dataclasses import replace +import logging +import tkinter as tk +from typing import TYPE_CHECKING +from unittest.mock import MagicMock + +from gatelock._config import LockConfig +from gatelock._prompts import PromptYield +from gatelock.tests.conftest import BOTH, build_loop + +if TYPE_CHECKING: + import pytest + + from gatelock._recovery import RecoveryLoop + from gatelock._surfaces import SurfaceSet + +REASON = "grab failed: another application has grab" + + +def _with_yield(recovery: RecoveryLoop, policy: MagicMock) -> MagicMock: + """Swap the loop's prompt policy for a mock.""" + recovery._parts = replace(recovery._parts, prompt_yield=policy) + return policy + + +def _policy(*, yielding: bool = False, was_yielding: bool = False) -> MagicMock: + policy = MagicMock(spec=PromptYield) + policy.yielding = yielding + policy.grab_held.return_value = was_yielding + return policy + + +def _tick_lifts(surfaces: SurfaceSet) -> int: + """How many lift() calls the surfaces have seen.""" + return sum(s.window.lift.call_count for s in surfaces._surfaces.values()) + + +class TestBlockedGrab: + """A failed re-grab is explained and handed to the yield policy.""" + + def test_failure_text_is_logged_once( + self, + loop: tuple[RecoveryLoop, MagicMock, SurfaceSet], + mock_root: MagicMock, + caplog: pytest.LogCaptureFixture, + ) -> None: + """Tk's own message names the cause; the same cause is not repeated.""" + recovery, enumerator, _ = loop + enumerator.scan.return_value = BOTH + mock_root.grab_current.return_value = None + mock_root.grab_set_global.side_effect = tk.TclError(REASON) + with caplog.at_level(logging.WARNING): + recovery.tick() + recovery.tick() + assert caplog.text.count(REASON) == 1 + + def test_a_different_cause_is_logged_again( + self, + loop: tuple[RecoveryLoop, MagicMock, SurfaceSet], + mock_root: MagicMock, + caplog: pytest.LogCaptureFixture, + ) -> None: + """ "Window not viewable" after "another application has grab" is news.""" + recovery, enumerator, _ = loop + enumerator.scan.return_value = BOTH + mock_root.grab_current.return_value = None + with caplog.at_level(logging.WARNING): + mock_root.grab_set_global.side_effect = tk.TclError(REASON) + recovery.tick() + mock_root.grab_set_global.side_effect = tk.TclError("not viewable") + recovery.tick() + assert "not viewable" in caplog.text + + def test_failure_feeds_the_policy( + self, loop: tuple[RecoveryLoop, MagicMock, SurfaceSet], mock_root: MagicMock + ) -> None: + """The reason reaches the policy that times the block.""" + recovery, enumerator, _ = loop + policy = _with_yield(recovery, _policy()) + enumerator.scan.return_value = BOTH + mock_root.grab_current.return_value = None + mock_root.grab_set_global.side_effect = tk.TclError(REASON) + recovery.tick() + policy.grab_blocked.assert_called_once_with(REASON) + + +class TestYielding: + """While a prompt is being yielded to, the lock stops covering it.""" + + def test_no_lift_while_yielding( + self, loop: tuple[RecoveryLoop, MagicMock, SurfaceSet], mock_root: MagicMock + ) -> None: + """enforce() still places and maps surfaces, but does not raise them.""" + recovery, enumerator, surfaces = loop + enumerator.scan.return_value = BOTH + mock_root.grab_current.return_value = None + mock_root.grab_set_global.side_effect = tk.TclError(REASON) + recovery.tick() # creates the surfaces (a create lifts once) + _with_yield(recovery, _policy(yielding=True)) + before = _tick_lifts(surfaces) + recovery.tick() + assert _tick_lifts(surfaces) == before + + def test_lifts_normally_when_not_yielding( + self, loop: tuple[RecoveryLoop, MagicMock, SurfaceSet], mock_root: MagicMock + ) -> None: + """Anything that is not a known prompt keeps today's behaviour.""" + recovery, enumerator, surfaces = loop + enumerator.scan.return_value = BOTH + recovery.tick() + before = _tick_lifts(surfaces) + _with_yield(recovery, _policy(yielding=False)) + recovery.tick() + assert _tick_lifts(surfaces) > before + + def test_settles_last( + self, loop: tuple[RecoveryLoop, MagicMock, SurfaceSet], mock_root: MagicMock + ) -> None: + """The prompt is re-raised after everything else the tick did.""" + recovery, enumerator, _ = loop + enumerator.scan.return_value = BOTH + policy = _with_yield(recovery, _policy(yielding=True)) + seen: list[str] = [] + + def grab_current() -> MagicMock: + seen.append("grab") + return mock_root + + def settle() -> None: + seen.append("settle") + + policy.settle.side_effect = settle + mock_root.grab_current.side_effect = grab_current + recovery.tick() + assert seen == ["grab", "settle"] + + def test_grab_retaken_lifts_the_lock_again( + self, loop: tuple[RecoveryLoop, MagicMock, SurfaceSet], mock_root: MagicMock + ) -> None: + """The prompt closed, the grab came back: the lock covers everything again.""" + recovery, enumerator, surfaces = loop + enumerator.scan.return_value = BOTH + recovery.tick() + _with_yield(recovery, _policy(yielding=True, was_yielding=True)) + before = _tick_lifts(surfaces) + mock_root.grab_current.return_value = None # lost, then re-taken + recovery.tick() + assert _tick_lifts(surfaces) > before + mock_root.lift.assert_called() + + def test_grab_held_is_reported_when_nothing_was_lost( + self, loop: tuple[RecoveryLoop, MagicMock, SurfaceSet], mock_root: MagicMock + ) -> None: + """Holding the grab already (the arming loop took it) also ends the yield.""" + recovery, enumerator, _ = loop + policy = _with_yield(recovery, _policy()) + enumerator.scan.return_value = BOTH + mock_root.grab_current.return_value = mock_root + recovery.tick() + policy.grab_held.assert_called_once_with() + + +def test_surface_set_enforce_without_lift(mock_root: MagicMock) -> None: + """``enforce(lift=False)`` corrects drift but never restacks.""" + recovery, enumerator, surfaces = build_loop(mock_root, LockConfig(mode="hard")) + enumerator.scan.return_value = BOTH + recovery.tick() + window = surfaces._surfaces["DP-0"].window + window.withdraw() # unmapped: must be remapped even when not lifting + window.lift.reset_mock() + assert surfaces.enforce(lift=False) == ("DP-0",) + window.deiconify.assert_called() + window.lift.assert_not_called() + + +def test_holds_grab_is_the_loops_own_question( + loop: tuple[RecoveryLoop, MagicMock, SurfaceSet], mock_root: MagicMock +) -> None: + """The public probe embedders and verification harnesses use.""" + recovery, _, _ = loop + mock_root.grab_current.return_value = mock_root + assert recovery.holds_grab() is True + mock_root.grab_current.return_value = None + assert recovery.holds_grab() is False + + +def test_raise_all_leaves_the_root_alone_without_a_backdrop( + mock_root: MagicMock, +) -> None: + """A soft, grab-less lock has no backdrop root to lift.""" + config = LockConfig(mode="soft") + _, _, surfaces = build_loop(mock_root, config) + surfaces.raise_all() + mock_root.lift.assert_not_called() diff --git a/gatelock/gatelock/tests/test_stepback.py b/gatelock/gatelock/tests/test_stepback.py new file mode 100644 index 0000000..8ac2072 --- /dev/null +++ b/gatelock/gatelock/tests/test_stepback.py @@ -0,0 +1,120 @@ +# Copyright (c) 2026 Krzysztof Rudnicki +"""Tests for a weaker armed lock stepping back for a stronger one. + +The trigger is the whole safety argument, so most tests are about what must NOT +trigger it: equal rank, lower rank, a dead claim, a stronger-ranked app that +locks more weakly, a loop the app stopped on purpose. See ``_stepback.py``. +""" + +from __future__ import annotations + +import logging +from typing import TYPE_CHECKING +from unittest.mock import MagicMock + +import pytest + +from gatelock._arbiter import RANK_DIET_GUARD, RANK_SCREEN_LOCKER +from gatelock._config import LockConfig +from gatelock._window import LockWindow +from gatelock.tests.conftest import make_arbiter + +if TYPE_CHECKING: + from pathlib import Path + + from gatelock._arbiter import Arbiter + +RANK_MID = 150 + + +def _tick(window: LockWindow) -> None: + window._stepback._tick() + + +class TestTrigger: + """Only a strictly stronger, live, at-least-as-firm claim makes it step back.""" + + def test_stronger_live_claim_suspends( + self, + armed: tuple[LockWindow, Arbiter], + mock_root: MagicMock, + tmp_path: Path, + caplog: pytest.LogCaptureFixture, + ) -> None: + """Hide, release the grab, drop the holder; the claim stays published.""" + window, arbiter = armed + strong = make_arbiter(tmp_path, "screen_locker", RANK_SCREEN_LOCKER) + windows = [s.window for s in window.surfaces._surfaces.values()] + with caplog.at_level(logging.WARNING): + _tick(window) + assert window._stepback.suspended + assert "steps back" in caplog.text + assert "screen_locker" in caplog.text + mock_root.grab_release.assert_called_once_with() + mock_root.withdraw.assert_called_once_with() + assert windows + assert all(w.withdraw.called for w in windows) + assert not window.recovery.running + assert not arbiter.holds_screen + assert arbiter.claim.grab_held is False + assert arbiter.claim.instance_id in { + c.instance_id for c in strong.live_claims() + } + strong.release() + + @pytest.mark.parametrize( + ("app", "rank"), + [("twin", RANK_MID), ("diet_guard", RANK_DIET_GUARD)], + ) + def test_equal_or_lower_rank_never_does( + self, armed: tuple[LockWindow, Arbiter], tmp_path: Path, app: str, rank: int + ) -> None: + """Rank must be strictly higher.""" + window, _ = armed + other = make_arbiter(tmp_path, app, rank) + _tick(window) + assert not window._stepback.suspended + other.release() + + def test_dead_claim_never_does( + self, armed: tuple[LockWindow, Arbiter], tmp_path: Path + ) -> None: + """A stale claim file (its owner died, flock gone) is no reason.""" + window, _ = armed + ghost = make_arbiter(tmp_path, "screen_locker", RANK_SCREEN_LOCKER) + assert ghost._claim_handle is not None + ghost._claim_handle.close() # the kernel's view of a dead owner + _tick(window) + assert not window._stepback.suspended + + def test_a_weaker_locking_app_never_does( + self, armed: tuple[LockWindow, Arbiter], tmp_path: Path + ) -> None: + """Rank must never weaken the lock: no grab, no VT -> no step back.""" + window, _ = armed + soft = make_arbiter( + tmp_path, "alarm", RANK_SCREEN_LOCKER, grab="none", vt=False + ) + _tick(window) + assert not window._stepback.suspended + soft.release() + + def test_a_loop_the_app_stopped_is_left_alone( + self, armed: tuple[LockWindow, Arbiter], tmp_path: Path + ) -> None: + """Study mode stops recovery on purpose; that is not ours to suspend.""" + window, _ = armed + window.recovery.stop() + strong = make_arbiter(tmp_path, "screen_locker", RANK_SCREEN_LOCKER) + _tick(window) + assert not window._stepback.suspended + strong.release() + + def test_without_an_arbiter_it_is_inert(self, mock_root: MagicMock) -> None: + """No arbitration, nothing to step back for.""" + window = LockWindow(mock_root, LockConfig(mode="hard"), MagicMock()) + window.setup() + window.grab_input() + mock_root.after.reset_mock() + window._stepback.start() + assert window._stepback._job is None diff --git a/gatelock/gatelock/tests/test_stepback_parts.py b/gatelock/gatelock/tests/test_stepback_parts.py new file mode 100644 index 0000000..7b2aaea --- /dev/null +++ b/gatelock/gatelock/tests/test_stepback_parts.py @@ -0,0 +1,109 @@ +# Copyright (c) 2026 Krzysztof Rudnicki +"""Tests for the small pieces step-back is built from. + +The startup grace in ``wait_for_turn``, ``Arbiter.drop_holder`` and +``SurfaceSet.set_visible``. The controller itself is in ``test_stepback.py``. +""" + +from __future__ import annotations + +from typing import TYPE_CHECKING +from unittest.mock import MagicMock, call, patch + +from gatelock import RANK_DIET_GUARD, RANK_SCREEN_LOCKER, Arbiter +from gatelock._config import LockConfig +from gatelock._queue import STARTUP_GRACE_SECONDS, Timebase, wait_for_turn +from gatelock.tests.conftest import DP0, build_loop, make_arbiter + +if TYPE_CHECKING: + from pathlib import Path + + +class TestStartupGrace: + """Simultaneous starters must see each other before anyone arms.""" + + def test_the_real_constant_is_set(self) -> None: + """Not zero in production (the test fixture zeroes the patched copy).""" + assert STARTUP_GRACE_SECONDS == 1.5 + + def test_waits_before_the_first_check_and_sees_a_late_publisher( + self, tmp_path: Path + ) -> None: + """A stronger claim published DURING the grace is queued behind.""" + weak = make_arbiter(tmp_path, "diet_guard", RANK_DIET_GUARD) + strong: list[Arbiter] = [] + slept: list[float] = [] + + def sleep(seconds: float) -> None: + slept.append(seconds) + if not strong: + strong.append( + make_arbiter(tmp_path, "screen_locker", RANK_SCREEN_LOCKER) + ) + elif len(slept) > 2: + strong[0].release() + + clock = [0.0] + base = Timebase(sleep=sleep, now=lambda: clock[0]) + with patch("gatelock._queue.STARTUP_GRACE_SECONDS", 1.5): + result = wait_for_turn(weak, timebase=base, poll=0.5) + assert slept[0] == 1.5 + assert result.queued + assert result.blocked_by == ("screen_locker",) + + def test_zero_grace_does_not_sleep(self, tmp_path: Path) -> None: + """The fixture's zero skips the sleep entirely.""" + slept: list[float] = [] + base = Timebase(sleep=slept.append, now=lambda: 0.0) + wait_for_turn( + make_arbiter(tmp_path, "diet_guard", RANK_DIET_GUARD), timebase=base + ) + assert slept == [] + + +class TestDropHolder: + """Dropping the holder lock leaves the claim, and is idempotent.""" + + def test_drop_keeps_the_claim_and_frees_the_screen(self, tmp_path: Path) -> None: + """Another app can take the holder lock; ours is still queued-behind.""" + ours = make_arbiter(tmp_path, "diet_guard", RANK_DIET_GUARD) + other = make_arbiter(tmp_path, "screen_locker", RANK_SCREEN_LOCKER) + assert ours.acquire_holder() + assert not other.acquire_holder() + ours.drop_holder() + ours.drop_holder() + assert not ours.holds_screen + assert ours.claim.instance_id in {c.instance_id for c in other.live_claims()} + assert other.acquire_holder() + + +class TestSetVisible: + """Hide, and show in creation order.""" + + def test_hide_then_show(self, mock_root: MagicMock) -> None: + """Withdrawn windows come back override-redirect BEFORE they map.""" + _, enumerator, surfaces = build_loop(mock_root, LockConfig(mode="hard")) + enumerator.scan.return_value = MagicMock(live=(DP0,)) + surfaces.apply(enumerator.scan.return_value) + window = surfaces._surfaces["DP-0"].window + window.reset_mock() + surfaces.set_visible(visible=False) + window.withdraw.assert_called_once_with() + window.reset_mock() + surfaces.set_visible(visible=True) + names = [c[0] for c in window.method_calls] + assert names.index("overrideredirect") < names.index("deiconify") + assert names[-1] == "lift" + assert call.overrideredirect(boolean=True) in window.method_calls + + def test_soft_mode_is_not_forced_override_redirect( + self, mock_root: MagicMock + ) -> None: + """A managed lock must stay managed.""" + _, enumerator, surfaces = build_loop(mock_root, LockConfig(mode="soft")) + enumerator.scan.return_value = MagicMock(live=(DP0,)) + surfaces.apply(enumerator.scan.return_value) + window = surfaces._surfaces["DP-0"].window + window.reset_mock() + surfaces.set_visible(visible=True) + window.overrideredirect.assert_not_called() diff --git a/gatelock/gatelock/tests/test_stepback_rearm.py b/gatelock/gatelock/tests/test_stepback_rearm.py new file mode 100644 index 0000000..af5f844 --- /dev/null +++ b/gatelock/gatelock/tests/test_stepback_rearm.py @@ -0,0 +1,165 @@ +# Copyright (c) 2026 Krzysztof Rudnicki +"""Re-arming after a step-back, and the failure paths around it. + +Split from ``test_stepback.py`` (250-line cap); the ``armed`` fixture is in +``conftest.py``. +""" + +from __future__ import annotations + +import logging +import tkinter as tk +from typing import TYPE_CHECKING +from unittest.mock import MagicMock, patch + +from gatelock import _stepback +from gatelock._arbiter import RANK_SCREEN_LOCKER +from gatelock._config import GrabPolicy, LockConfig +from gatelock._window import LockWindow +from gatelock.tests.conftest import make_arbiter + +if TYPE_CHECKING: + from pathlib import Path + + import pytest + + from gatelock._arbiter import Arbiter + +RANK_MID = 150 + + +def _tick(window: LockWindow) -> None: + window._stepback._tick() + + +class TestRearm: + """It comes back, fully, once nothing qualifying is live.""" + + def _suspended( + self, armed: tuple[LockWindow, Arbiter], tmp_path: Path + ) -> tuple[LockWindow, Arbiter, Arbiter]: + window, arbiter = armed + strong = make_arbiter(tmp_path, "screen_locker", RANK_SCREEN_LOCKER) + _tick(window) + assert window._stepback.suspended + return window, arbiter, strong + + def test_stays_hidden_while_the_stronger_is_live( + self, armed: tuple[LockWindow, Arbiter], tmp_path: Path + ) -> None: + """Windowless until the stronger one is gone.""" + window, _, strong = self._suspended(armed, tmp_path) + _tick(window) + assert window._stepback.suspended + strong.release() + + def test_rearms_fully_when_it_is_gone( + self, + armed: tuple[LockWindow, Arbiter], + mock_root: MagicMock, + tmp_path: Path, + caplog: pytest.LogCaptureFixture, + ) -> None: + """VT, backdrop, surfaces, grab, holder and the loops all come back.""" + window, arbiter, strong = self._suspended(armed, tmp_path) + strong.release() + mock_root.grab_set_global.reset_mock() + with ( + patch("gatelock._stepback.disable_vt_switching") as vt, + caplog.at_level(logging.WARNING), + ): + _tick(window) + assert not window._stepback.suspended + assert "re-arms" in caplog.text + vt.assert_called_once_with() + mock_root.deiconify.assert_called_once_with() + mock_root.grab_set_global.assert_called_once_with() + assert arbiter.holds_screen + assert window.recovery.running + for surface in window.surfaces._surfaces.values(): + surface.window.deiconify.assert_called() + + def test_without_vt_policy_it_leaves_vt_alone( + self, mock_root: MagicMock, tmp_path: Path + ) -> None: + """A lock that never disabled VT does not start now.""" + arbiter = make_arbiter(tmp_path, "leetcode_guard", RANK_MID, vt=False) + grab = GrabPolicy(disable_vt=False) + config = LockConfig(mode="hard", grab=grab, rank=RANK_MID) + window = LockWindow(mock_root, config, MagicMock(), arbiter=arbiter) + window.setup() + window.grab_input() + strong = make_arbiter(tmp_path, "screen_locker", RANK_SCREEN_LOCKER, vt=False) + _tick(window) + strong.release() + with patch("gatelock._stepback.disable_vt_switching") as vt: + _tick(window) + assert not window._stepback.suspended + vt.assert_not_called() + window.close() + + def test_gives_up_waiting_after_the_deadline( + self, + armed: tuple[LockWindow, Arbiter], + tmp_path: Path, + caplog: pytest.LogCaptureFixture, + ) -> None: + """A stuck stronger claim cannot keep the machine unlocked forever.""" + window, _, strong = self._suspended(armed, tmp_path) + step = window._stepback + step._since = step._clock() - _stepback.QUEUE_DEADLINE_SECONDS - 1 + with caplog.at_level(logging.ERROR): + _tick(window) + assert not step.suspended + assert "gave up" in caplog.text + strong.release() + + +class TestRobustness: + """A failing step must not leave the lock half-down.""" + + def test_tk_errors_are_logged_and_the_rest_still_happens( + self, + armed: tuple[LockWindow, Arbiter], + mock_root: MagicMock, + tmp_path: Path, + caplog: pytest.LogCaptureFixture, + ) -> None: + """A failed grab release is logged; hiding and the claim still follow.""" + window, arbiter = armed + mock_root.grab_release.side_effect = tk.TclError("no grab") + strong = make_arbiter(tmp_path, "screen_locker", RANK_SCREEN_LOCKER) + with caplog.at_level(logging.WARNING): + _tick(window) + assert "grab release failed" in caplog.text + assert window._stepback.suspended + assert not arbiter.holds_screen + strong.release() + + def test_a_raise_does_not_end_the_checking( + self, armed: tuple[LockWindow, Arbiter], caplog: pytest.LogCaptureFixture + ) -> None: + """The reschedule is in ``finally``; one bad check is just logged.""" + window, _ = armed + with ( + patch("gatelock._stepback.outranking", side_effect=ValueError("boom")), + caplog.at_level(logging.ERROR), + ): + _tick(window) + assert "step-back check raised" in caplog.text + assert window._stepback._job is not None + + def test_close_stops_the_checking(self, armed: tuple[LockWindow, Arbiter]) -> None: + """No timer outlives the window.""" + window, _ = armed + window.close() + assert window._stepback._job is None + + def test_stop_survives_a_dead_root( + self, armed: tuple[LockWindow, Arbiter], mock_root: MagicMock + ) -> None: + """Cancelling on a destroyed root is logged, not raised.""" + window, _ = armed + mock_root.after_cancel.side_effect = tk.TclError("gone") + window._stepback.stop() + assert window._stepback._job is None diff --git a/gatelock/pyproject.toml b/gatelock/pyproject.toml index e801762..124c5a7 100644 --- a/gatelock/pyproject.toml +++ b/gatelock/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta" [project] name = "gatelock" -version = "0.12.0" +version = "0.13.0" description = "Shared per-output lock-window + HMAC log-integrity backend for blocking-overlay apps" requires-python = ">=3.14" dependencies = [] # pure stdlib — tkinter is bundled with Python diff --git a/gatelock/scripts/locker_child.py b/gatelock/scripts/locker_child.py new file mode 100755 index 0000000..3764037 --- /dev/null +++ b/gatelock/scripts/locker_child.py @@ -0,0 +1,101 @@ +#!/usr/bin/env python3 +# Copyright (c) 2026 Krzysztof Rudnicki +# pylint: disable=missing-function-docstring +"""One stand-in locker for ``verify_all_lockers.py``: the real startup sequence. + +Mirrors what leetcode_guard / screen_locker / diet_guard / book_guard / +focus_guard do: ``Arbiter`` -> ``publish`` -> ``wait_for_turn`` (windowless) -> +``acquire_holder`` -> ``LockWindow`` -> ``setup`` -> ``grab_input`` -> mainloop, +with the real app name and rank. Its only UI is one button; clicking it ends +the lock, like a solve does. Everything is reported on stdout, one line each: +``ARMED``, ``BTN x y`` (button centre, root coordinates), ``GRAB 0|1``, +``CLICKED``. + +Usage: ``locker_child.py `` +""" + +from __future__ import annotations + +import sys +import tkinter as tk + +from gatelock import ( + Arbiter, + GateRoot, + LockConfig, + LockWindow, + wait_for_turn, +) + +REPORT_MS = 250 + + +class _Child: + """The hooks object, as each real locker is.""" + + def __init__(self, app: str, rank: int, poll: float) -> None: + self.app = app + self.button: tk.Button | None = None + config = LockConfig(mode="hard", app_name=app, rank=rank) + arbiter = Arbiter( + app, + rank, + grab=config.resolved_grab(), + disable_vt=False, + ) + arbiter.publish() + wait_for_turn(arbiter, poll=poll) + arbiter.acquire_holder() + self.root = GateRoot(className=app) + self.lock = LockWindow(self.root, config, self, arbiter=arbiter) + self.lock.setup() + self.say("ARMED") + self.lock.grab_input() + self.root.after(REPORT_MS, self.report) + + def say(self, text: str) -> None: + sys.stdout.write(f"{text}\n") + sys.stdout.flush() + + def report(self) -> None: + self.say(f"GRAB {int(self.lock.recovery.holds_grab())}") + button = self.button + if button is not None and button.winfo_ismapped(): + x = button.winfo_rootx() + button.winfo_width() // 2 + y = button.winfo_rooty() + button.winfo_height() // 2 + self.say(f"BTN {x} {y}") + self.root.after(REPORT_MS, self.report) + + def clicked(self) -> None: + self.say("CLICKED") + self.lock.close() + + # -- LockWindowHooks --------------------------------------------------- + def build_surface(self, parent: tk.Misc, surface: object) -> None: + del surface + self.button = tk.Button(parent, text=self.app, command=self.clicked) + self.button.place(relx=0.5, rely=0.5, width=240, height=90, anchor="center") + + def teardown_surface(self, surface: object) -> None: + del surface + self.button = None + + def on_focus_ready(self, surface: object) -> None: + del surface + + def on_callback_error(self) -> None: + self.say("ERROR callback") + + def on_close(self) -> None: + return + + +def main(argv: list[str]) -> int: + """Arm one locker and run its mainloop.""" + child = _Child(argv[0], int(argv[1]), float(argv[2])) + child.lock.run() + return 0 + + +if __name__ == "__main__": + sys.exit(main(sys.argv[1:])) diff --git a/gatelock/scripts/prompt_grab_helper.py b/gatelock/scripts/prompt_grab_helper.py new file mode 100755 index 0000000..9770205 --- /dev/null +++ b/gatelock/scripts/prompt_grab_helper.py @@ -0,0 +1,73 @@ +#!/usr/bin/env python3 +# Copyright (c) 2026 Krzysztof Rudnicki +# pylint: disable=import-error +"""A second X client that holds the keyboard+pointer grab, like a credential prompt. + +Started by ``verify_prompt_yield.py`` on a private display. Maps one window with +the given ``WM_CLASS`` (optionally inside an override-redirect frame, the way a +window manager reparents clients), takes ``XGrabKeyboard`` + ``XGrabPointer``, +prints ``READY``, then waits on stdin: ``release`` ungrabs and unmaps (prints +``RELEASED``), EOF exits. + +Usage: ``prompt_grab_helper.py [--framed]`` +""" + +from __future__ import annotations + +import sys +import time + +from Xlib import X, display + +OWNER_EVENTS = True +VIEWABLE_TRIES = 50 + + +def main(argv: list[str]) -> int: + """Grab, announce, wait for ``release``; returns a process exit status.""" + wm_class = argv[0] + xdisplay = display.Display() + root = xdisplay.screen().root + depth = xdisplay.screen().root_depth + top = window = root.create_window( + 10, 10, 400, 200, 0, depth, X.InputOutput, X.CopyFromParent + ) + if "--framed" in argv: + top = root.create_window( + 10, 10, 400, 200, 0, depth, X.InputOutput, X.CopyFromParent, + override_redirect=1, + ) # fmt: skip + window.reparent(top, 0, 0) + window.set_wm_class(wm_class, wm_class.capitalize()) + window.map() + top.map() + xdisplay.sync() + # A window manager maps asynchronously; grabbing a not-yet-viewable window + # fails with GrabNotViewable. + for _ in range(VIEWABLE_TRIES): + if window.get_attributes().map_state == X.IsViewable: + break + time.sleep(0.1) + modes = (X.GrabModeAsync, X.GrabModeAsync) + keyboard = window.grab_keyboard(OWNER_EVENTS, *modes, X.CurrentTime) + pointer = window.grab_pointer( + OWNER_EVENTS, X.ButtonPressMask, *modes, X.NONE, X.NONE, X.CurrentTime + ) + if keyboard or pointer: + sys.stdout.write(f"grab failed keyboard={keyboard} pointer={pointer}\n") + return 2 + sys.stdout.write("READY\n") + sys.stdout.flush() + for line in sys.stdin: + if line.strip() == "release": + xdisplay.ungrab_keyboard(X.CurrentTime) + xdisplay.ungrab_pointer(X.CurrentTime) + top.unmap() + xdisplay.sync() + sys.stdout.write("RELEASED\n") + sys.stdout.flush() + return 0 + + +if __name__ == "__main__": + sys.exit(main(sys.argv[1:])) diff --git a/gatelock/scripts/verify_all_lockers.py b/gatelock/scripts/verify_all_lockers.py new file mode 100755 index 0000000..7625903 --- /dev/null +++ b/gatelock/scripts/verify_all_lockers.py @@ -0,0 +1,249 @@ +#!/usr/bin/env python3 +# Copyright (c) 2026 Krzysztof Rudnicki +# pylint: disable=missing-function-docstring,consider-using-with,import-error +r"""All lockers at once, on a real X server: one lock at a time, each clickable. + +Starts the five lockers that fire together at login (real app names and ranks, +the real ``Arbiter`` + ``wait_for_turn`` startup, one process each, a private +``GATELOCK_RUNTIME_DIR``) with a ``gcr-prompter`` grab helper started first, as +on 2026-11-11. Asserts: never two locks on screen; once the prompt lets go the +lone lock holds the grab; an XTEST click on its button fires its callback; when +it closes the next one comes up the same way. The queue poll is 0.5 s (real: +2 s) to fit the 60 s cap. Argument: launch order -- mixed, strong-first, +weak-first, or late-strong (screen_locker starts 3 s after the rest, once a +weaker lock holds the grab). Xvfb only (XTEST); refuses ``:0``:: + + PYTHONPATH=. xvfb-run -a -s "-screen 0 1600x1200x24" \ + python3 scripts/verify_all_lockers.py +""" + +from __future__ import annotations + +import os +from pathlib import Path +import re +import select +import subprocess +import sys +import tempfile +import time +from typing import TYPE_CHECKING, Final + +from Xlib import X, display, error +from Xlib.ext import xtest + +if TYPE_CHECKING: + from collections.abc import Callable + +HERE: Final = Path(__file__).parent +LOCKERS: Final = { # app -> rank, as in each locker's own constants + "screen_locker": 200, + "leetcode_guard": 150, + "book_guard": 125, + "focus_guard": 120, + "diet_guard": 100, +} +_RANKED: Final = tuple(sorted(LOCKERS, key=lambda a: -LOCKERS[a])) +ORDERS: Final = { # launch order: how the five processes race to publish + "mixed": tuple(_RANKED[i] for i in (4, 1, 3, 0, 2)), + "strong-first": _RANKED, + "weak-first": _RANKED[::-1], + "late-strong": tuple(sorted(set(LOCKERS) - {"screen_locker"})), +} +LATE: Final = {"late-strong": ("screen_locker", 3.0, 2.0)} # app, start s, hold s +POLL_S, STEP_S, PROMPT_HOLD_S, SHOW_S, GRAB_S, CLICK_S = 0.5, 0.1, 6.0, 12.0, 8.0, 5.0 +SETTLE_S = 1.0 + + +def _say(message: str) -> None: + sys.stdout.write(f"{message}\n") + + +class _Fleet: + """The child processes and what they have reported so far.""" + + def __init__(self, env: dict[str, str], launch: tuple[str, ...]) -> None: + self.env = env + self.procs: dict[str, subprocess.Popen[bytes]] = {} + self.buffers = dict.fromkeys(LOCKERS, b"") + self.grab = dict.fromkeys(LOCKERS, 0) + self.button: dict[str, tuple[int, int]] = {} + self.clicked: set[str] = set() + self.armed: set[str] = set() + for app in launch: + self.launch(app) + + def launch(self, app: str) -> None: + args = [sys.executable, str(HERE / "locker_child.py"), app] + args += [str(LOCKERS[app]), str(POLL_S)] + self.procs[app] = subprocess.Popen(args, stdout=subprocess.PIPE, env=self.env) + + def drain(self) -> None: + fds = {p.stdout.fileno(): a for a, p in self.procs.items() if p.stdout} + for fd in select.select(list(fds), [], [], 0)[0]: + app = fds[fd] + self.buffers[app] += os.read(fd, 65536) + *lines, self.buffers[app] = self.buffers[app].split(b"\n") + for raw in lines: + self.note(app, raw.decode()) + + def note(self, app: str, line: str) -> None: + if line == "ARMED": + self.armed.add(app) + elif line == "CLICKED": + self.clicked.add(app) + elif m := re.fullmatch(r"GRAB (\d)", line): + self.grab[app] = int(m[1]) + elif m := re.fullmatch(r"BTN (\d+) (\d+)", line): + self.button[app] = (int(m[1]), int(m[2])) + elif line.startswith("ERROR"): + _say(f"FAIL {app} reported {line}") + + def stop(self) -> None: + for proc in self.procs.values(): + proc.terminate() + for proc in self.procs.values(): + proc.wait(timeout=10) + + +class _Screen: + """Which locks are on screen, and a synthetic mouse.""" + + def __init__(self) -> None: + self.display = display.Display() + self.root = self.display.screen().root + + def shown(self) -> set[str]: + apps: set[str] = set() + try: + for child in self.root.query_tree().children: + if child.get_attributes().map_state != X.IsViewable: + continue + for win in (child, *child.query_tree().children): + found = win.get_wm_class() + if found and found[1].lower() in LOCKERS: + apps.add(found[1].lower()) + except error.XError as exc: # a lock closed mid-walk: next sample + _say(f"note: window vanished mid-sample ({exc})") + return apps + + def click(self, x: int, y: int) -> None: + # Approach from off-target: Tk learns which widget is under the pointer + # from Enter/Motion events, and the previous lock's button sat at this + # very spot, so a motion to the same coordinates would produce none. + for px, py in ((x - 30, y - 30), (x, y)): + xtest.fake_input(self.display, X.MotionNotify, x=px, y=py) + self.display.sync() + time.sleep(0.15) + xtest.fake_input(self.display, X.ButtonPress, 1) + xtest.fake_input(self.display, X.ButtonRelease, 1) + self.display.sync() + + +class _Run: + """Pumps the fleet and samples the screen; remembers any double lock.""" + + def __init__(self, fleet: _Fleet, screen: _Screen) -> None: + self.fleet = fleet + self.screen = screen + self.doubles: list[list[str]] = [] + + def pump(self, seconds: float, until: Callable[[], bool] | None = None) -> bool: + deadline = time.monotonic() + seconds + while time.monotonic() < deadline: + self.fleet.drain() + shown = self.screen.shown() + if len(shown) > 1: + self.doubles.append(sorted(shown)) + if until is not None and until(): + return True + time.sleep(STEP_S) + return until() if until is not None else False + + def lone(self, seen: list[str]) -> str | None: + """Wait for exactly one new lock on screen; return its app.""" + got = self.pump( + SHOW_S, + lambda: len(shown := self.screen.shown()) == 1 and not shown & set(seen), + ) + return next(iter(self.screen.shown())) if got else None + + +def _drive(run: _Run, check: Callable[[object, str], None]) -> list[str]: + """Click each lock that appears, in turn; returns the order seen.""" + order: list[str] = [] + fleet = run.fleet + for _ in LOCKERS: + app = run.lone(order) + check(app, f"lock #{len(order) + 1} came up alone ({app})") + if not app: + break + order.append(app) + ready = run.pump(GRAB_S, lambda a=app: fleet.grab[a] == 1 and a in fleet.button) + check(ready, f"{app}: holds the grab and shows its button") + if ready: + run.pump(SETTLE_S) + run.screen.click(*fleet.button[app]) + check( + run.pump(CLICK_S, lambda a=app: a in fleet.clicked), + f"{app}: XTEST click fired the callback", + ) + return order + + +def main(mode: str) -> int: + """Run the scenario on a private display; 0 when every check held.""" + if os.environ.get("DISPLAY", "").split(".")[0] in {"", ":0"}: + _say("refusing to run: DISPLAY must be a private X server, never :0") + return 2 + if mode not in ORDERS: + _say(f"usage: verify_all_lockers.py [{'|'.join(ORDERS)}]") + return 2 + env = {**os.environ, "GATELOCK_RUNTIME_DIR": tempfile.mkdtemp(prefix="gl-lockers-")} + failures: list[str] = [] + + def check(ok: object, label: str) -> None: + _say(f"{'PASS' if ok else 'FAIL'} {label}") + if not ok: + failures.append(label) + + helper = subprocess.Popen( + [sys.executable, str(HERE / "prompt_grab_helper.py"), "gcr-prompter"], + stdin=subprocess.PIPE, + stdout=subprocess.PIPE, + text=True, + ) + check( + helper.stdout is not None and helper.stdout.readline().strip() == "READY", + "prompt holds the grab", + ) + fleet = _Fleet(env, ORDERS[mode]) + run = _Run(fleet, _Screen()) + try: + run.pump(LATE[mode][2] if mode in LATE else PROMPT_HOLD_S) + check(not run.doubles, "while the prompt held the grab: never two locks") + if helper.stdin: + helper.stdin.write("release\n") + helper.stdin.flush() + if mode in LATE: + late, at, hold = LATE[mode] + run.pump(at - hold) + check("leetcode_guard" in fleet.armed, "a weaker lock is armed first") + fleet.launch(late) + gone = run.pump(SHOW_S, lambda: "leetcode_guard" not in run.screen.shown()) + check(gone, "the weaker lock stepped back") + order = _drive(run, check) + check(not run.doubles, f"never two locks on screen (saw {run.doubles[:3]})") + _say(f"order seen: {order}") + _say("rank order" if order == list(_RANKED) else "NOT rank order") + finally: + fleet.stop() + if helper.stdin: + helper.stdin.close() + helper.wait(timeout=10) + _say("OK" if not failures else f"{len(failures)} FAILED") + return 1 if failures else 0 + + +if __name__ == "__main__": + sys.exit(main(sys.argv[1] if len(sys.argv) > 1 else "mixed")) diff --git a/gatelock/scripts/verify_grab_responsive.py b/gatelock/scripts/verify_grab_responsive.py new file mode 100755 index 0000000..1504dc2 --- /dev/null +++ b/gatelock/scripts/verify_grab_responsive.py @@ -0,0 +1,113 @@ +#!/usr/bin/env python3 +# Copyright (c) 2026 Krzysztof Rudnicki +# pylint: disable=consider-using-with,import-error +r"""Prove the lock's Tk loop stays responsive while a foreign client holds the grab. + +Tk's ``grab -global`` blocks the whole event loop for ~1.0 s on every FAILED +attempt, and the lock retries every 200 ms plus once a tick, so under a held +grab (2026-11-11: the keyring prompt) the loop was starved: no repaint, no +clicks, rows and buttons without text. + +``prompt_grab_helper.py`` holds the grab (class ``steam_game``: not a prompt, +so no yielding muddies the measurement). A real ``mainloop`` runs for 5 s with +a 50 ms ``after`` heartbeat; the worst gap between beats must stay under +300 ms. After the helper lets go the lock must still take the grab. + + PYTHONPATH=. xvfb-run -a -s "-screen 0 1600x1200x24" \ + python3 scripts/verify_grab_responsive.py +""" + +from __future__ import annotations + +import os +from pathlib import Path +import select +import subprocess +import sys +import time +from unittest.mock import MagicMock + +from gatelock import GateRoot, LockConfig, LockWindow +from gatelock._config import GrabPolicy + +HELPER = Path(__file__).with_name("prompt_grab_helper.py") +BEAT_MS, RUN_S, MAX_GAP_S, WAIT_S = 50, 5.0, 0.3, 8.0 + + +def _say(message: str) -> None: + sys.stdout.write(f"{message}\n") + + +def _line(proc: subprocess.Popen[str]) -> str: + out = proc.stdout + ready = select.select([out], [], [], WAIT_S)[0] if out else [] + return out.readline().strip() if ready and out else "" + + +def _worst_gap(root: GateRoot) -> float: + """Run the real mainloop for RUN_S; return the largest heartbeat gap.""" + last = [time.monotonic()] + worst = [0.0] + + def beat() -> None: + now = time.monotonic() + worst[0] = max(worst[0], now - last[0]) + last[0] = now + root.after(BEAT_MS, beat) + + root.after(BEAT_MS, beat) + root.after(int(RUN_S * 1000), root.quit) + root.mainloop() + return worst[0] + + +def main() -> int: + """Hold a foreign grab, measure the loop, then release; 0 when all held.""" + if os.environ.get("DISPLAY", "").split(".")[0] in {"", ":0"}: + _say("refusing to run: DISPLAY must be a private X server, never :0") + return 2 + helper = subprocess.Popen( + [sys.executable, str(HELPER), "steam_game"], + stdin=subprocess.PIPE, + stdout=subprocess.PIPE, + text=True, + ) + failed = 0 + if _line(helper) != "READY": + _say("FAIL helper did not take the grab") + return 1 + root = GateRoot() + config = LockConfig(mode="hard", grab=GrabPolicy(disable_vt=False, retry_ms=200)) + lock = LockWindow(root, config, MagicMock()) + try: + lock.setup() + lock.grab_input() + gap = _worst_gap(root) + ok = gap < MAX_GAP_S + _say(f"{'PASS' if ok else 'FAIL'} worst loop gap {gap:.3f}s (< {MAX_GAP_S}s)") + failed += not ok + grab_held = lock.recovery.holds_grab() + _say(f"{'FAIL' if grab_held else 'PASS'} helper still holds the grab") + failed += grab_held + if helper.stdin: + helper.stdin.write("release\n") + helper.stdin.flush() + _line(helper) + deadline = time.monotonic() + WAIT_S + while time.monotonic() < deadline and not lock.recovery.holds_grab(): + root.update() + time.sleep(0.05) + got = lock.recovery.holds_grab() + _say(f"{'PASS' if got else 'FAIL'} lock took the grab once it was free") + failed += not got + finally: + lock.close() + if helper.stdin: + helper.stdin.close() + helper.wait(timeout=WAIT_S) + _say("OK" if not failed else f"{failed} FAILED") + return 1 if failed else 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/gatelock/scripts/verify_prompt_yield.py b/gatelock/scripts/verify_prompt_yield.py new file mode 100755 index 0000000..ad668d1 --- /dev/null +++ b/gatelock/scripts/verify_prompt_yield.py @@ -0,0 +1,250 @@ +#!/usr/bin/env python3 +# Copyright (c) 2026 Krzysztof Rudnicki +# pylint: disable=missing-function-docstring,consider-using-with,import-error +r"""Prove, on a real X server, that the lock yields to a system credential prompt. + +No unit test can: in the suite the Tk root is a MagicMock, so the grab never +fails and stacking is never real. The 2026-11-11 incident was exactly that. + +``prompt_grab_helper.py`` (a second X client, so Tk really answers "another +application has grab") maps a window with a chosen ``WM_CLASS`` and takes the +keyboard+pointer grab; a ``LockWindow`` then arms over it. Cases: ``prompt`` +(gcr-prompter must end up above backdrop and surfaces, stay there, and the lock +must retake the grab and the top once the helper lets go), ``framed`` (client +inside an override-redirect frame, like i3: the top-level ancestor is raised), +``game`` (steam_game is not allowlisted: lock stays on top), ``i3`` (like +``prompt`` under a real i3 with a private runtime dir). + +Run one case on a throwaway display (never ``:0``):: + + PYTHONPATH=. xvfb-run -a -s "-screen 0 1600x1200x24" \ + python3 scripts/verify_prompt_yield.py prompt +""" + +from __future__ import annotations + +import _tkinter +import os +from pathlib import Path +import select +import subprocess +import sys +import tempfile +import time +import tkinter as tk +from typing import TYPE_CHECKING, Final +from unittest.mock import MagicMock + +from Xlib import X, display + +from gatelock import GateRoot, LockConfig, LockWindow +from gatelock._config import GrabPolicy + +if TYPE_CHECKING: + from collections.abc import Callable + +HELPER: Final = Path(__file__).with_name("prompt_grab_helper.py") +CASES: Final = { # case -> (WM_CLASS instance, framed) + "prompt": ("gcr-prompter", False), + "framed": ("gcr-prompter", True), + "game": ("steam_game", False), + "i3": ("gcr-prompter", False), +} +STEP_S, WAIT_S = 0.05, 8.0 + + +def _say(message: str) -> None: + sys.stdout.write(f"{message}\n") + + +class _Helper: + """The grab-holding subprocess, as seen from the checking side.""" + + def __init__(self, wm_class: str, *, framed: bool) -> None: + args = [sys.executable, str(HELPER), wm_class, *(["--framed"] * framed)] + self.proc = subprocess.Popen( + args, stdin=subprocess.PIPE, stdout=subprocess.PIPE, text=True + ) + self._expect("READY") + + def _expect(self, word: str) -> None: + out = self.proc.stdout + ready = select.select([out], [], [], WAIT_S)[0] if out else [] + line = out.readline().strip() if ready and out else "" + if line != word: + msg = f"helper said {line!r}, wanted {word!r}" + raise RuntimeError(msg) + + def release(self) -> None: + if self.proc.stdin: + self.proc.stdin.write("release\n") + self.proc.stdin.flush() + self._expect("RELEASED") + + def stop(self) -> None: + if self.proc.stdin: + self.proc.stdin.close() + self.proc.wait(timeout=WAIT_S) + + +class _Stack: + """Real stacking order, read over the harness's own X connection.""" + + def __init__(self, lock_tops: list[int]) -> None: + self.display = display.Display() + self.root = self.display.screen().root + self.lock_ids = [self.top_of(i) for i in lock_tops] + self.prompt = 0 + + def top_of(self, window_id: int) -> int: + win = self.display.create_resource_object("window", window_id) + while (parent := win.query_tree().parent).id != self.root.id: + win = parent + return int(win.id) + + def order(self) -> list[int]: + return [w.id for w in self.root.query_tree().children] + + def viewable(self, window_id: int) -> bool: + win = self.display.create_resource_object("window", window_id) + return bool(win.get_attributes().map_state == X.IsViewable) + + def find(self, wm_class: str) -> int | None: + """The root child that is, or contains, the helper's window.""" + for child in self.root.query_tree().children: + for win in (child, *child.query_tree().children): + found = win.get_wm_class() + if found and found[0] == wm_class: + self.prompt = int(child.id) + return self.prompt + return None + + def covered(self) -> bool: + order = self.order() + return all(order.index(i) > order.index(self.prompt) for i in self.lock_ids) + + def on_top(self) -> bool: + order = self.order() + return all(order.index(self.prompt) > order.index(i) for i in self.lock_ids) + + def lock_topmost(self) -> bool: + """The topmost *viewable* root child belongs to the lock.""" + top = next(i for i in reversed(self.order()) if self.viewable(i)) + return top in self.lock_ids + + +def _pump(root: tk.Tk, seconds: float, until: Callable[[], bool] | None = None) -> bool: + """Run Tk's event loop for ``seconds`` or until ``until`` holds. + + One event at a time: ``update()`` never returns while the lock's grab + retries (each blocks Tk for ~1 s) keep coming due. + """ + deadline = time.monotonic() + seconds + while time.monotonic() < deadline: + root.tk.dooneevent(_tkinter.DONT_WAIT) + if until is not None and until(): + return True + time.sleep(STEP_S) + return until() if until is not None else False + + +def _start_i3() -> subprocess.Popen[bytes]: + """Start i3 on the private display, with a private runtime dir.""" + home = Path(tempfile.mkdtemp(prefix="gl-i3-")) + (home / "config").write_text("font pango:monospace 8\n", encoding="utf-8") + # No inherited I3SOCK: this i3 must never touch the live session's socket. + env = {k: v for k, v in os.environ.items() if k != "I3SOCK"} + env["XDG_RUNTIME_DIR"] = str(home) + proc = subprocess.Popen(["/usr/bin/i3", "-c", str(home / "config")], env=env) + probe = display.Display() + atom = probe.intern_atom("I3_SOCKET_PATH") + deadline = time.monotonic() + WAIT_S + while time.monotonic() < deadline: + if probe.screen().root.get_full_property(atom, X.AnyPropertyType): + return proc + time.sleep(0.1) + proc.kill() + msg = "i3 never came up" + raise RuntimeError(msg) + + +def _checks(case: str, root: tk.Tk, lock: LockWindow, helper: _Helper) -> list[str]: + """Drive the armed lock through one scenario; returns failed labels.""" + wm_class = CASES[case][0] + failures: list[str] = [] + + def check(ok: object, label: str) -> None: + _say(f"{'PASS' if ok else 'FAIL'} {label}") + if not ok: + failures.append(label) + + _pump(root, 0.3) + tops: list[tk.Wm] = [root] + tops += [w for w in root.winfo_children() if isinstance(w, tk.Toplevel)] + seen = _Stack([int(t.wm_frame(), 16) for t in tops]) + prompt = seen.find(wm_class) + check(prompt is not None, "helper window found on the root") + if prompt is None: + return failures + _pump(root, 0.7) + check(seen.covered(), "t<1s: lock starts above the prompt (incident)") + check(not lock.recovery.holds_grab(), "t<1s: the helper holds the grab") + yielded = _pump(root, WAIT_S, seen.on_top) + if case == "game": + check(not yielded, "steam_game: lock stays on top (no yield)") + _pump(root, 2.5) + check(seen.covered(), "steam_game: still covered 2.5s later") + else: + check(yielded, "prompt raised above backdrop and every surface") + _pump(root, 2.5) # several enforce ticks + check(seen.on_top(), "prompt stays on top across ticks") + check(not lock.recovery.holds_grab(), "lock does not hold the grab yet") + check(all(seen.viewable(i) for i in seen.lock_ids), "lock still mapped") + helper.release() + got = _pump(root, WAIT_S, lock.recovery.holds_grab) + check(got, "after the helper let go, the lock took the grab") + _pump(root, 1.5) + check(seen.lock_topmost(), "lock is back on top of everything viewable") + return failures + + +def _run_case(case: str) -> list[str]: + """Arm a lock over the helper's grab, run the scenario, tear it all down.""" + wm_class, framed = CASES[case] + wm = _start_i3() if case == "i3" else None + helper = lock = None + try: + helper = _Helper(wm_class, framed=framed) + root = GateRoot() + config = LockConfig( + mode="hard", grab=GrabPolicy(disable_vt=False, retry_ms=200) + ) + lock = LockWindow(root, config, MagicMock()) + lock.setup() + lock.grab_input() + return _checks(case, root, lock, helper) + finally: + if lock is not None: + lock.close() + if helper is not None: + helper.stop() + if wm is not None: + wm.terminate() + + +def main(argv: list[str]) -> int: + """Run the named case on a private display; 0 when every check held.""" + xdisplay = os.environ.get("DISPLAY", "") + if not xdisplay or xdisplay.split(".")[0] == ":0": + _say("refusing to run: DISPLAY must be a private X server, never :0") + return 2 + if len(argv) != 1 or argv[0] not in CASES: + _say(f"usage: verify_prompt_yield.py {{{'|'.join(CASES)}}}") + return 2 + failed = _run_case(argv[0]) + _say("OK" if not failed else f"{len(failed)} FAILED") + return 1 if failed else 0 + + +if __name__ == "__main__": + sys.exit(main(sys.argv[1:]))