jakub
a951a96edc
Wait for both convoy cargo phases and use actual magazine stock. Resolve empty primary weapon ammunition from muzzle metadata and recover cleanly from failed initial settings saves. Update native fixtures, scoped cleanup guards, packaging evidence and validation documentation. Make reviewed fixture tests checkout-relative. All repository checks pass against the isolated staged tree; native resupply completion and multiplayer qualification remain outstanding.
323 lines
14 KiB
Python
323 lines
14 KiB
Python
#!/usr/bin/env python3
|
|
"""Narrow post-run cleanup of adopted device helpers in the one disposable prefix.
|
|
|
|
Library only: importing or executing this file sends no signals. Call only while
|
|
holding .erri-workbench.lock, after normal wineserver/Gamescope shutdown. This is
|
|
not a general Wine cleanup tool. Environment values are never returned/logged.
|
|
"""
|
|
from dataclasses import dataclass
|
|
from pathlib import Path
|
|
import os
|
|
import posixpath
|
|
import select
|
|
import signal
|
|
import stat
|
|
import time
|
|
|
|
COMPAT = Path('/tmp/erri-headless-compat')
|
|
INSTALLED_PROTON = Path.home()/'.local/share/Steam/steamapps/common/Proton - Experimental/proton'
|
|
PREFIX = b'/tmp/erri-headless-compat/pfx'
|
|
DEVICE_ARGV = (b'C:\\windows\\system32\\winedevice.exe',)
|
|
SYSTEMD_ARGV = (b'/usr/lib/systemd/systemd', b'--user')
|
|
|
|
|
|
class CleanupRefused(ValueError):
|
|
def __init__(self, message, report=None):
|
|
super().__init__(message)
|
|
self.report = report or {}
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class Process:
|
|
pid: int
|
|
uid: int
|
|
ppid: int
|
|
start: int
|
|
state: str
|
|
comm: str
|
|
argv: tuple
|
|
exe: str | None
|
|
executable_id: tuple | None
|
|
prefix: bytes | None = None
|
|
|
|
def identity(self):
|
|
# State may legitimately change between running and sleeping.
|
|
return (self.pid, self.uid, self.ppid, self.start, self.comm, self.argv,
|
|
self.exe, self.executable_id)
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class Device:
|
|
process: Process
|
|
parent: Process
|
|
|
|
def identity(self):
|
|
return self.process.identity(), self.parent.identity()
|
|
|
|
def public(self):
|
|
p = self.process
|
|
return {'pid': p.pid, 'ppid': p.ppid, 'state': p.state,
|
|
'exebasename': 'wine64-preloader', 'start_ticks': p.start,
|
|
'role': 'adopted-winedevice'}
|
|
|
|
|
|
def prefix_kind(value):
|
|
if value is None:
|
|
return 'other'
|
|
if value.rstrip(b'/') == PREFIX:
|
|
return 'exact'
|
|
# Never resolve untrusted environment paths or follow their symlinks.
|
|
# A lexical spelling of this prefix is ambiguous and cannot be killed.
|
|
if posixpath.normpath(value) == PREFIX:
|
|
return 'alias'
|
|
return 'other'
|
|
|
|
|
|
def user_systemd_argv(argv):
|
|
if argv == SYSTEMD_ARGV:
|
|
return True
|
|
# User-systemd preserves its identity across reexec and appends its state
|
|
# descriptor. Accept only the observed single ASCII decimal FD argument.
|
|
if len(argv) != 3 or argv[:2] != SYSTEMD_ARGV:
|
|
return False
|
|
argument = argv[2]
|
|
if not argument.startswith(b'--deserialize='):
|
|
return False
|
|
descriptor = argument[len(b'--deserialize='):]
|
|
return (descriptor.isdigit() and len(descriptor) <= 10
|
|
and int(descriptor) <= 2147483647)
|
|
|
|
|
|
def native_metadata_proves_unrelated(p):
|
|
names = [p.comm.lower(), *(os.fsdecode(a).replace('\\', '/').rsplit('/', 1)[-1].lower() for a in p.argv)]
|
|
if p.exe:
|
|
names.append(Path(p.exe).name.lower())
|
|
if any(n.endswith('.exe') or n.startswith(('wine', 'proton', 'gamescope', 'armareforger')) or n in {'reaper', 'pressure-vessel-wrap', 'pv-bwrap'} for n in names):
|
|
return False
|
|
if (p.comm == 'systemd' or p.exe == '/usr/lib/systemd/systemd'
|
|
or (p.argv and p.argv[0] == SYSTEMD_ARGV[0])):
|
|
return (p.comm == 'systemd' and p.ppid == 1
|
|
and user_systemd_argv(p.argv)
|
|
and p.exe in {None, '/usr/lib/systemd/systemd'})
|
|
if p.exe and p.exe.startswith('/'):
|
|
return True
|
|
if p.comm == '(sd-pam)' and p.argv == (b'(sd-pam)',):
|
|
return True
|
|
# Same narrow native foreground signature as headless_game's preflight.
|
|
if p.comm == 'apache2' and p.argv == (b'apache2', b'-DFOREGROUND'):
|
|
return True
|
|
# The user's GnuPG smart-card helper is native and may be nondumpable.
|
|
if p.comm == 'scdaemon' and p.argv == (b'scdaemon', b'--multi-server'):
|
|
return True
|
|
# Nondumpable desktop services still expose argv and kernel comm. Match
|
|
# both (Linux comm truncates at 15 bytes), after refusing runtime markers
|
|
# in every argument above. This does not authorize any signal: it only
|
|
# skips an unreadable environment for positively native metadata.
|
|
if p.argv and p.argv[0].startswith(b'/'):
|
|
basename = os.fsdecode(p.argv[0]).rsplit('/', 1)[-1]
|
|
if p.comm and p.comm in {basename, basename[:15]}:
|
|
return True
|
|
return False
|
|
|
|
|
|
def prove_device(p, parent, uid, expected_exe, executable_id):
|
|
if (p.uid != uid or p.state in {'Z', 'X'} or prefix_kind(p.prefix) != 'exact'
|
|
or p.comm != 'winedevice.exe' or p.argv != DEVICE_ARGV
|
|
or p.exe != expected_exe or p.executable_id != executable_id):
|
|
raise CleanupRefused(f'PID {p.pid} is not the exact adopted device helper; left untouched')
|
|
if (parent.pid != p.ppid or parent.uid != uid or parent.ppid != 1
|
|
or parent.state in {'Z', 'X'} or parent.comm != 'systemd'
|
|
or not user_systemd_argv(parent.argv)
|
|
or parent.exe not in {None, '/usr/lib/systemd/systemd'}):
|
|
raise CleanupRefused(f'PID {p.pid} does not have the proven user-systemd parent; left untouched')
|
|
# The observed user-systemd is nondumpable: an unreadable exe is accepted
|
|
# only with all exact parent metadata above. A readable wrong exe is refused.
|
|
return Device(p, parent)
|
|
|
|
|
|
class ProcFS:
|
|
def __init__(self, expected_exe, executable_id, root=Path('/proc'), uid=None, self_pid=None):
|
|
self.root = root
|
|
self.uid = os.getuid() if uid is None else uid
|
|
self.self_pid = os.getpid() if self_pid is None else self_pid
|
|
self.expected_exe = str(expected_exe)
|
|
self.executable_id = executable_id
|
|
|
|
def metadata(self, pid):
|
|
p = self.root / str(pid)
|
|
uid = p.stat().st_uid
|
|
text = (p/'stat').read_text()
|
|
fields = text[text.rfind(')') + 2:].split()
|
|
state, ppid, start = fields[0], int(fields[1]), int(fields[19])
|
|
comm = (p/'comm').read_text().strip()
|
|
argv = tuple(a for a in (p/'cmdline').read_bytes().split(b'\0') if a)
|
|
try:
|
|
exe = os.readlink(p/'exe')
|
|
info = (p/'exe').stat()
|
|
executable_id = (info.st_dev, info.st_ino)
|
|
except PermissionError:
|
|
exe, executable_id = None, None
|
|
# Detect PID reuse while reading the separate proc records.
|
|
final = (p/'stat').read_text()
|
|
after = final[final.rfind(')') + 2:].split()
|
|
if int(after[19]) != start or int(after[1]) != ppid:
|
|
raise CleanupRefused(f'PID {pid} changed while reading metadata; left untouched')
|
|
return Process(pid, uid, ppid, start, state, comm, argv, exe, executable_id)
|
|
|
|
def devices(self):
|
|
found = []
|
|
for entry in self.root.iterdir():
|
|
if not entry.name.isdigit() or int(entry.name) == self.self_pid:
|
|
continue # library caller is this process, never another PID
|
|
pid = int(entry.name)
|
|
try:
|
|
if entry.stat().st_uid != self.uid:
|
|
continue
|
|
p = self.metadata(pid)
|
|
if p.state in {'Z', 'X'}:
|
|
continue # already exited, not a signal target
|
|
try:
|
|
values = [v[11:] for v in (entry/'environ').read_bytes().split(b'\0') if v.startswith(b'WINEPREFIX=')]
|
|
except PermissionError:
|
|
if native_metadata_proves_unrelated(p):
|
|
continue
|
|
raise CleanupRefused(f'Cannot establish prefix ownership for PID {pid}; left untouched')
|
|
kinds = [prefix_kind(value) for value in values]
|
|
if not any(k in {'exact', 'alias'} for k in kinds):
|
|
continue
|
|
if len(values) != 1 or kinds[0] != 'exact':
|
|
raise CleanupRefused(f'PID {pid} has ambiguous private-prefix spelling; left untouched')
|
|
again = self.metadata(pid)
|
|
if again.identity() != p.identity():
|
|
raise CleanupRefused(f'PID {pid} changed during prefix inspection; left untouched')
|
|
p = Process(**{**p.__dict__, 'prefix': values[0]})
|
|
parent = self.metadata(p.ppid)
|
|
found.append(prove_device(p, parent, self.uid, self.expected_exe, self.executable_id))
|
|
except (FileNotFoundError, ProcessLookupError):
|
|
continue
|
|
except (PermissionError, OSError) as error:
|
|
raise CleanupRefused(f'Cannot verify private-prefix PID {pid}: {type(error).__name__}; left untouched') from None
|
|
return sorted(found, key=lambda d: d.process.pid)
|
|
|
|
|
|
class Pidfds:
|
|
def open(self, pid):
|
|
return os.pidfd_open(pid)
|
|
|
|
def close(self, fd):
|
|
os.close(fd)
|
|
|
|
def send(self, fd, sig):
|
|
signal.pidfd_send_signal(fd, sig)
|
|
|
|
def wait(self, handles, timeout):
|
|
poller = select.poll()
|
|
by_fd = {fd: pid for pid, fd in handles.items()}
|
|
for fd in by_fd:
|
|
poller.register(fd, select.POLLIN)
|
|
finished = set()
|
|
until = time.monotonic() + timeout
|
|
while len(finished) < len(handles):
|
|
remaining = max(0, until-time.monotonic())
|
|
events = poller.poll(int(remaining*1000))
|
|
for fd, flags in events:
|
|
if flags & (select.POLLIN | select.POLLHUP):
|
|
finished.add(by_fd[fd])
|
|
poller.unregister(fd)
|
|
elif flags & (select.POLLERR | select.POLLNVAL):
|
|
raise CleanupRefused('A pinned process handle became unverifiable')
|
|
if remaining <= 0 or time.monotonic() >= until:
|
|
break
|
|
return finished
|
|
|
|
|
|
def _cleanup(inspector, signals):
|
|
"""Testable algorithm; production entry below always supplies real guards."""
|
|
initial = {d.process.pid: d for d in inspector.devices()}
|
|
report = {'scope': str(COMPAT/'pfx'), 'candidates': [d.public() for d in initial.values()],
|
|
'term_sent': [], 'kill_sent': [], 'cleared': False}
|
|
handles = {}
|
|
|
|
def census():
|
|
current = {d.process.pid: d for d in inspector.devices()}
|
|
for pid, candidate in current.items():
|
|
if pid not in initial or candidate.identity() != initial[pid].identity():
|
|
raise CleanupRefused(f'Private-prefix process identity changed (PID {pid}); remaining processes left untouched')
|
|
return current
|
|
|
|
try:
|
|
# Open every handle only after the complete prefix census proves there
|
|
# are no active/unknown processes. Before signalling, verify it again.
|
|
for pid in initial:
|
|
try:
|
|
handles[pid] = signals.open(pid)
|
|
except ProcessLookupError:
|
|
pass
|
|
for pid, fd in handles.items():
|
|
current = census()
|
|
if pid in signals.wait({pid: fd}, 0):
|
|
continue
|
|
if pid not in current:
|
|
raise CleanupRefused(f'Pinned PID {pid} lost its proven prefix identity; left untouched')
|
|
try:
|
|
signals.send(fd, signal.SIGTERM)
|
|
report['term_sent'].append(pid)
|
|
except ProcessLookupError:
|
|
pass # pinned original exited, never retry via its numeric PID
|
|
exited = signals.wait(handles, 3.0)
|
|
for pid, fd in handles.items():
|
|
if pid in exited:
|
|
continue
|
|
current = census()
|
|
if pid in signals.wait({pid: fd}, 0):
|
|
continue
|
|
if pid not in current:
|
|
raise CleanupRefused(f'Pinned PID {pid} lost its proven prefix identity; left untouched')
|
|
try:
|
|
signals.send(fd, signal.SIGKILL)
|
|
report['kill_sent'].append(pid)
|
|
except ProcessLookupError:
|
|
pass
|
|
signals.wait(handles, 1.0)
|
|
if census():
|
|
raise CleanupRefused('Some pinned private-prefix helpers remain alive; no broader cleanup attempted')
|
|
report['cleared'] = True
|
|
return report
|
|
except (CleanupRefused, OSError) as error:
|
|
raise CleanupRefused(str(error), report) from None
|
|
finally:
|
|
for fd in handles.values():
|
|
signals.close(fd)
|
|
|
|
|
|
def _inspector(compat, proton):
|
|
compat, proton = Path(compat), Path(proton)
|
|
if compat != COMPAT or compat.resolve() != COMPAT or (compat/'pfx').resolve() != COMPAT/'pfx':
|
|
raise CleanupRefused('Cleanup is restricted to the canonical disposable /tmp/erri-headless-compat/pfx')
|
|
for path in (compat, compat/'pfx'):
|
|
if not path.is_dir() or path.stat().st_uid != os.getuid():
|
|
raise CleanupRefused('Private prefix must be an existing directory owned by this user')
|
|
expected = proton.parent/'files/lib/wine/i386-unix/wine64-preloader'
|
|
if proton != INSTALLED_PROTON or not proton.is_file() or not expected.is_file() or expected.resolve() != expected:
|
|
raise CleanupRefused('Verified installed Proton device-helper executable required')
|
|
info = expected.stat()
|
|
if not stat.S_ISREG(info.st_mode):
|
|
raise CleanupRefused('Proton device-helper executable is not a regular file')
|
|
if not hasattr(os, 'pidfd_open') or not hasattr(signal, 'pidfd_send_signal'):
|
|
raise CleanupRefused('PID-pinned signalling is unavailable; no PID-only fallback')
|
|
return ProcFS(expected, (info.st_dev, info.st_ino))
|
|
|
|
|
|
def inspect_private_prefix(compat, proton):
|
|
"""Read-only proof; refuses if anything except exact adopted helpers remains."""
|
|
return [d.public() for d in _inspector(compat, proton).devices()]
|
|
|
|
|
|
def cleanup_private_prefix(compat, proton):
|
|
"""After ordinary shutdown, with the shared prefix lock still held.
|
|
|
|
Never call as startup recovery for a busy prefix. The caller owns the
|
|
disposed test session; any other live process makes this fail closed.
|
|
"""
|
|
return _cleanup(_inspector(compat, proton), Pidfds())
|