fix(launcher): address hide-client review findings

- associate target windows with the selected prefix through the guest PID,
  X11 _NET_WM_PID, NSpid and the loader socket; fail closed when ownership is
  unknown;
- let --once run without either opt-in gate, matching the documented
  diagnostic;
- route SIGHUP through the restore handler;
- cache non-empty WM_CLASS results per window and prune stale entries;
- cover unreadable /proc, host PID mapping and prefix ownership in tests.
This commit is contained in:
3AvisD committed 2026-10-22 12:32:03 -06:00
1 parent 899576cbbc
commit 521527eb21
2 files changed
+167 -20

No files matched your search

+62 -5
View File
@@ -88,6 +88,43 @@ class HideClientInGameTests(unittest.TestCase):
with patch.object(Path, 'read_bytes', read):
self.assertIsNone(hide_in_game.game_running(self.prefix, self.proc))
def test_unreadable_process_directory_is_unknown(self):
class UnreadableProc:
def iterdir(self):
raise OSError()
self.assertIsNone(hide_in_game.game_running(self.prefix, UnreadableProc()))
def test_host_socket_for_guest_uses_nspid(self):
host = self.proc / '999'
host.mkdir()
(host / 'status').write_text('Name:\tCrBrowserMain\nNSpid:\t999\t5877\n')
(host / 'environ').write_bytes(self.environment(self.prefix))
self.assertEqual(
hide_in_game.host_socket_for_guest(5877, self.proc),
os.fsencode(self.prefix / hide_in_game.SOCK_NAME),
)
def test_window_belongs_to_prefix(self):
socket = os.fsencode(self.prefix / hide_in_game.SOCK_NAME)
with patch.object(hide_in_game, 'window_pid', return_value=5877), \
patch.object(hide_in_game, 'host_socket_for_guest', return_value=socket):
self.assertTrue(hide_in_game.window_belongs_to_prefix('0x10', self.prefix))
self.assertFalse(hide_in_game.window_belongs_to_prefix(
'0x10', self.root / 'other'))
def test_once_runs_without_opt_in(self):
output = io.StringIO()
with patch.dict(os.environ, {
'RCLOL_HIDE_CLIENT_IN_GAME': '',
'RCLOL_CLOSE_CLIENT_IN_GAME': '',
}), patch.object(hide_in_game, 'target_prefix',
return_value=self.prefix), \
patch.object(hide_in_game, 'game_running', return_value=True), \
patch.object(hide_in_game.sys, 'stdout', output):
self.assertEqual(hide_in_game.main(['--once']), 0)
self.assertIn('match=running', output.getvalue())
def test_target_windows_ignores_game_class(self):
with patch.object(hide_in_game, 'xprop_root',
return_value=['0x10', '0x20', '0x30']), \
@@ -96,8 +133,28 @@ class HideClientInGameTests(unittest.TestCase):
'wm_class(string) = "leagueclientux"',
'wm_class(string) = "leagueoflegends"',
'wm_class(string) = "riot client"',
]):
self.assertEqual(hide_in_game.target_windows(), [0x10, 0x30])
]), \
patch.object(hide_in_game, 'window_belongs_to_prefix', return_value=True):
self.assertEqual(hide_in_game.target_windows(self.prefix), [0x10, 0x30])
def test_target_windows_requires_this_prefix(self):
with patch.object(hide_in_game, 'xprop_root',
return_value=['0x10', '0x20']), \
patch.object(hide_in_game, 'window_class',
return_value='wm_class(string) = "leagueclientux"'), \
patch.object(hide_in_game, 'window_belongs_to_prefix',
side_effect=[True, False]):
self.assertEqual(hide_in_game.target_windows(self.prefix), [0x10])
def test_target_windows_caches_non_empty_classes(self):
cache = {}
with patch.object(hide_in_game, 'xprop_root', return_value=['0x10']), \
patch.object(hide_in_game, 'window_class',
return_value='wm_class(string) = "leagueclientux"') as classes, \
patch.object(hide_in_game, 'window_belongs_to_prefix', return_value=True):
self.assertEqual(hide_in_game.target_windows(self.prefix, cache), [0x10])
self.assertEqual(hide_in_game.target_windows(self.prefix, cache), [0x10])
self.assertEqual(classes.call_count, 1)
def test_hides_then_shows_on_match_end(self):
states = iter([True, True, False])
@@ -105,7 +162,7 @@ class HideClientInGameTests(unittest.TestCase):
output = io.StringIO()
watcher = hide_in_game.Watcher(
self.prefix, x11,
window_source=lambda: [0xabc],
window_source=lambda _prefix, _cache: [0xabc],
state_source=lambda _prefix: next(states),
sleep=lambda _seconds: None,
out=output,
@@ -123,7 +180,7 @@ class HideClientInGameTests(unittest.TestCase):
output = io.StringIO()
watcher = hide_in_game.Watcher(
self.prefix, x11,
window_source=lambda: [0xdef],
window_source=lambda _prefix, _cache: [0xdef],
state_source=lambda _prefix: next(states),
sleep=lambda _seconds: None,
out=output,
@@ -137,7 +194,7 @@ class HideClientInGameTests(unittest.TestCase):
x11 = FakeX11()
watcher = hide_in_game.Watcher(
self.prefix, x11,
window_source=lambda: [0xabc],
window_source=lambda _prefix, _cache: [0xabc],
state_source=lambda _prefix: False,
sleep=lambda _seconds: None,
out=io.StringIO(),
+105 -15
View File
@@ -117,23 +117,112 @@ def xprop_root():
return re.findall(r'0x[0-9a-fA-F]+', listing)
def window_class(window):
def xprop_value(window, prop):
try:
prop = subprocess.check_output(
['xprop', '-id', window, 'WM_CLASS'], text=True, timeout=5
return subprocess.check_output(
['xprop', '-id', window, prop], text=True, timeout=5
)
except (OSError, subprocess.SubprocessError):
return ''
return prop.lower()
def target_windows():
"""Managed client windows only; the game window is never selected."""
def window_class(window):
return xprop_value(window, 'WM_CLASS').lower()
def window_pid(window):
match = re.search(r'=\s*(\d+)', xprop_value(window, '_NET_WM_PID'))
return int(match.group(1)) if match else None
def _socket_from_process(proc):
try:
environment = dict(
item.split(b'=', 1)
for item in (proc / 'environ').read_bytes().split(b'\0')
if b'=' in item
)
except (OSError, ValueError):
return None
return environment.get(SOCK_VARIABLE)
def host_socket_for_guest(guest_pid, proc_root=Path('/proc')):
"""Find the host process for a guest X11 PID and return its loader socket.
Darling's X11 window stores the guest PID in _NET_WM_PID, while the
process visible in /proc is the host loader. NSpid carries both IDs. The
direct /proc/<pid> case is checked too for a native process.
"""
direct = proc_root / str(guest_pid)
if direct.exists():
socket = _socket_from_process(direct)
if socket is not None:
return socket
sockets = set()
try:
entries = list(proc_root.iterdir())
except OSError:
return None
for proc in entries:
if not proc.name.isdigit():
continue
try:
status = (proc / 'status').read_text()
except OSError:
continue
for line in status.splitlines():
if not line.startswith('NSpid:'):
continue
if str(guest_pid) not in line.split()[1:]:
continue
socket = _socket_from_process(proc)
if socket is not None:
sockets.add(socket)
return sockets.pop() if len(sockets) == 1 else None
def window_belongs_to_prefix(window, prefix):
pid = window_pid(window)
if pid is None:
return False
socket = host_socket_for_guest(pid)
if socket is None:
return False
try:
return Path(os.fsdecode(socket)).parent.resolve() == (prefix / SOCK_NAME).parent.resolve()
except OSError:
return False
def target_windows(prefix, class_cache=None):
"""Managed client windows for this prefix; the game window is never selected.
WM_CLASS is cached only when non-empty. The cache is pruned when a window
leaves _NET_CLIENT_LIST; empty results are retried because a class can be
set after the window appears.
"""
current = []
for raw in xprop_root():
try:
current.append((int(raw, 16), raw))
except ValueError:
continue
current_ids = {window for window, _raw in current}
if class_cache is not None:
for stale in set(class_cache) - current_ids:
del class_cache[stale]
found = []
for window in xprop_root():
classes = window_class(window)
if any(name in classes for name in TARGET_CLASSES):
found.append(int(window, 16))
for window, raw in current:
classes = class_cache.get(window) if class_cache is not None else None
if not classes:
classes = window_class(raw)
if classes and class_cache is not None:
class_cache[window] = classes
if not classes or not any(name in classes for name in TARGET_CLASSES):
continue
if window_belongs_to_prefix(raw, prefix):
found.append(window)
return sorted(set(found))
@@ -179,6 +268,7 @@ class Watcher:
self.sleep = sleep
self.out = out
self.hidden = set()
self.class_cache = {}
self.reported_no_window = False
def _show_all(self, reason):
@@ -195,7 +285,7 @@ class Watcher:
self.hidden.clear()
def _hide_new(self):
windows = self.window_source()
windows = self.window_source(self.prefix, self.class_cache)
if not windows and not self.hidden and not self.reported_no_window:
print('hide-league-client-in-game: no managed client window; not hiding',
file=self.out, flush=True)
@@ -242,11 +332,11 @@ def main(argv=None):
help='one game observation and exit (no window changes)')
args = parser.parse_args(argv)
if os.environ.get('RCLOL_CLOSE_CLIENT_IN_GAME') == '1':
if not args.once and os.environ.get('RCLOL_CLOSE_CLIENT_IN_GAME') == '1':
print('hide-league-client-in-game: ignored; RCLOL_CLOSE_CLIENT_IN_GAME=1 takes precedence',
file=sys.stderr, flush=True)
return 0
if os.environ.get('RCLOL_HIDE_CLIENT_IN_GAME') != '1':
if not args.once and os.environ.get('RCLOL_HIDE_CLIENT_IN_GAME') != '1':
print('hide-league-client-in-game: disabled; set RCLOL_HIDE_CLIENT_IN_GAME=1',
flush=True)
return 0
@@ -277,8 +367,8 @@ def main(argv=None):
def stop(_signum, _frame):
raise KeyboardInterrupt
signal.signal(signal.SIGTERM, stop)
signal.signal(signal.SIGINT, stop)
for signum in (signal.SIGTERM, signal.SIGINT, signal.SIGHUP):
signal.signal(signum, stop)
print('hide-league-client-in-game: watching %s for a running match; '
'opt out with RCLOL_HIDE_CLIENT_IN_GAME=0' % prefix, flush=True)
try: