gatelock 0.13.0: locks stay clickable together
Some checks failed
dependency freshness / dependencies on newest stable (push) Waiting to run
file length / file-length (push) Waiting to run
md-naming / markdown naming convention (push) Waiting to run
palette drift / palette (push) Waiting to run
repo-contract / repo contract (push) Waiting to run
shared gate tests / gate unit tests (push) Waiting to run
gatelock pre-commit / pre-commit (push) Has been cancelled
gatelock tests / test (push) Has been cancelled

A keyring prompt (gcr-prompter) that grabbed input before the
leetcode lock armed left the lock visible but unclickable: every
click went to the prompt hidden under it. Each failed
grab_set_global also froze Tk ~1 s, so the lock barely repainted.

- Raise known credential prompts above the lock after 2 s of a
  blocked grab; games and other grabbers stay covered.
- Probe the grab on a side Xlib connection first so a blocked
  grab never stalls the Tk loop (worst gap 1.05 s -> 0.05 s).
- Log the Tk grab error and stop blaming our own claim.
- 1.5 s startup grace in wait_for_turn, and a weaker armed lock
  steps back (windowless, re-arms later) when a stronger one is
  live, so only one lock is ever on screen.
- Real-X checks: verify_prompt_yield, verify_grab_responsive,
  verify_all_lockers.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01WzxdWZbpxQZj41WpbQPuxB
This commit is contained in:
Krzysztof Rudnicki 2026-11-11 12:49:40 +01:00
parent 3084080d74
commit 2d341dd41e
35 changed files with 3398 additions and 63 deletions

View File

@ -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

View File

@ -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:

View File

@ -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()

View File

@ -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

View File

@ -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))

View File

@ -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,

View File

@ -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
)

View File

@ -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))

View File

@ -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

View File

@ -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)

View File

@ -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.

View File

@ -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

View File

@ -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."""

View File

@ -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()

View File

@ -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()

View File

@ -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()

View File

@ -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,
)

View File

@ -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()

View File

@ -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 <its own 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

View File

@ -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"

View File

@ -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

View File

@ -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

View File

@ -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

View File

@ -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"}

View File

@ -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")

View File

@ -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()

View File

@ -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

View File

@ -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()

View File

@ -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

View File

@ -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

101
gatelock/scripts/locker_child.py Executable file
View File

@ -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 <app> <rank> <queue_poll_s>``
"""
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:]))

View File

@ -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 <wm_class> [--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:]))

View File

@ -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"))

View File

@ -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())

View File

@ -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:]))