EscapistsReintroduced/tests/test_private_prefix_cleanup.py

"""Fake process/pidfd tests only. Never invoke native cleanup or send a signal."""
import dataclasses
import importlib.util
import signal
import sys
from pathlib import Path
import unittest
from unittest.mock import patch, MagicMock
ROOT=Path(__file__).resolve().parents[1]
spec=importlib.util.spec_from_file_location('private_prefix_cleanup',ROOT/'tools/private_prefix_cleanup.py')
m=importlib.util.module_from_spec(spec);sys.modules[spec.name]=m;spec.loader.exec_module(m)
UID=1000
EXE='/installed/Proton/files/lib/wine/i386-unix/wine64-preloader'
INODE=(2,44)
PARENT=m.Process(1670,UID,1,100,'S','systemd',m.SYSTEMD_ARGV,None,None)
def proc(pid=10,**changes):
 p=m.Process(pid,UID,PARENT.pid,1000+pid,'S','winedevice.exe',m.DEVICE_ARGV,EXE,INODE,m.PREFIX)
 return dataclasses.replace(p,**changes)
class Inspector:
 def __init__(self,processes=()):self.processes={p.pid:p for p in processes};self.parent=PARENT;self.reads=0
 def devices(self):
  self.reads+=1;result=[]
  for p in self.processes.values():
   if p.state in {'Z','X'} or m.prefix_kind(p.prefix)=='other':continue
   if m.prefix_kind(p.prefix)=='alias':raise m.CleanupRefused('ambiguous private-prefix spelling')
   result.append(m.prove_device(p,self.parent,UID,EXE,INODE))
  return result
class FakePidfds:
 def __init__(self,inspector,term_exits=True,kill_exits=True):
  self.inspector=inspector;self.term_exits=term_exits;self.kill_exits=kill_exits;self.opened={};self.closed=[];self.sent=[];self.waits=[];self.on_open=None;self.on_send=None
 def open(self,pid):
  if pid not in self.inspector.processes:raise ProcessLookupError()
  fd=pid+1000;self.opened[fd]=self.inspector.processes[pid]
  if self.on_open:self.on_open(pid)
  return fd
 def close(self,fd):self.closed.append(fd)
 def send(self,fd,sig):
  original=self.opened[fd];self.sent.append((original.pid,sig))
  if self.on_send:self.on_send(original.pid,sig)
  if (sig==signal.SIGTERM and self.term_exits) or (sig==signal.SIGKILL and self.kill_exits):
   current=self.inspector.processes.get(original.pid)
   if current and current.start==original.start:self.inspector.processes.pop(original.pid)
 def wait(self,handles,timeout):
  self.waits.append(timeout);finished=set()
  for pid,fd in handles.items():
   current=self.inspector.processes.get(pid);original=self.opened[fd]
   if not current or current.start!=original.start or current.state in {'Z','X'}:finished.add(pid)
  return finished

class PrivatePrefixCleanup(unittest.TestCase):
 def test_exact_prefix_and_observed_device(self):
  self.assertEqual(m.DEVICE_ARGV[0].hex(),'433a5c77696e646f77735c73797374656d33325c77696e656465766963652e657865')
  for suffix in (b'',b'/',b'//'):
   p=proc(prefix=m.PREFIX+suffix);self.assertEqual(m.prefix_kind(p.prefix),'exact');self.assertEqual(m.prove_device(p,PARENT,UID,EXE,INODE).process.pid,10)
  for other in (None,b'/other/pfx',m.PREFIX+b'-sibling',b'/tmp/another/pfx'):
   self.assertEqual(m.prefix_kind(other),'other')
  self.assertEqual(m.prefix_kind(m.PREFIX+b'/../pfx'),'alias')
 def test_identity_and_parent_refusals(self):
  mutations=[{'uid':UID+1},{'state':'Z'},{'argv':(*m.DEVICE_ARGV,b'extra')},{'argv':(b'Z:\\winedevice.exe',)},{'comm':'winedbg.exe'},{'exe':'/untrusted/wine64-preloader'},{'executable_id':(2,45)},{'prefix':b'/other/pfx'}]
  for mutation in mutations:
   with self.subTest(mutation=mutation),self.assertRaises(m.CleanupRefused):m.prove_device(proc(**mutation),PARENT,UID,EXE,INODE)
  for mutation in ({'uid':UID+1},{'pid':3},{'ppid':2},{'comm':'bash'},{'argv':(b'/usr/lib/systemd/systemd',)},{'exe':'/tmp/systemd'},{'state':'Z'}):
   with self.subTest(parent=mutation),self.assertRaises(m.CleanupRefused):m.prove_device(proc(),dataclasses.replace(PARENT,**mutation),UID,EXE,INODE)
  self.assertTrue(m.prove_device(proc(),dataclasses.replace(PARENT,exe='/usr/lib/systemd/systemd'),UID,EXE,INODE))
 def test_user_systemd_reexec_parent_and_unrelated_metadata(self):
  for descriptor in (b'0',b'10',b'2147483647'):
   parent=dataclasses.replace(PARENT,argv=(*m.SYSTEMD_ARGV,b'--deserialize='+descriptor))
   self.assertTrue(m.native_metadata_proves_unrelated(parent))
   self.assertEqual(m.prove_device(proc(),parent,UID,EXE,INODE).parent,parent)
   i=Inspector([proc()]);i.parent=parent;f=FakePidfds(i)
   self.assertTrue(m._cleanup(i,f)['cleared']);self.assertEqual(f.sent,[(10,signal.SIGTERM)])
  invalid=[(*m.SYSTEMD_ARGV,b'--deserialize='+value) for value in
           (b'',b'-1',b'+10',b'10 ',b' 10',b'10x',b'1.0',b'2147483648',b'99999999999')]
  invalid.extend([(*m.SYSTEMD_ARGV,b'--deserialize',b'10'),
                  (*m.SYSTEMD_ARGV,b'--deserialize=10',b'extra'),
                  (*m.SYSTEMD_ARGV,b'--deserialize=10',b'proton'),
                  (*m.SYSTEMD_ARGV,b'--unknown=10'),
                  (b'/tmp/systemd',b'--user',b'--deserialize=10')])
  for argv in invalid:
   parent=dataclasses.replace(PARENT,argv=argv)
   with self.subTest(argv=argv):
    self.assertFalse(m.user_systemd_argv(argv))
    self.assertFalse(m.native_metadata_proves_unrelated(parent))
    with self.assertRaises(m.CleanupRefused):m.prove_device(proc(),parent,UID,EXE,INODE)
  parent=dataclasses.replace(PARENT,argv=(*m.SYSTEMD_ARGV,b'--deserialize=10'))
  for mutation in ({'uid':UID+1},{'pid':3},{'ppid':2},{'comm':'bash'},
                   {'exe':'/tmp/systemd'},{'state':'Z'}):
   with self.subTest(mutation=mutation),self.assertRaises(m.CleanupRefused):
    m.prove_device(proc(),dataclasses.replace(parent,**mutation),UID,EXE,INODE)
 def test_reexec_parent_identity_change_before_signal_refused(self):
  for mutation in ({'argv':(*m.SYSTEMD_ARGV,b'--deserialize=11')},{'start':999}):
   i=Inspector([proc()]);i.parent=dataclasses.replace(PARENT,argv=(*m.SYSTEMD_ARGV,b'--deserialize=10'));f=FakePidfds(i)
   f.on_open=lambda pid:i.__setattr__('parent',dataclasses.replace(i.parent,**mutation))
   with self.subTest(mutation=mutation),self.assertRaises(m.CleanupRefused):m._cleanup(i,f)
   self.assertEqual(f.sent,[]);self.assertEqual(f.closed,[1010])
 def test_clean_term_ignores_other_prefix(self):
  i=Inspector([proc(10),proc(11,prefix=m.PREFIX+b'/'),proc(12,prefix=b'/other/pfx')]);f=FakePidfds(i)
  result=m._cleanup(i,f);self.assertTrue(result['cleared']);self.assertEqual(result['term_sent'],[10,11]);self.assertEqual(result['kill_sent'],[])
  self.assertEqual(set(i.processes),{12});self.assertEqual(set(f.closed),set(f.opened));self.assertEqual(max(f.waits),3.0)
  self.assertNotIn('argv',str(result));self.assertNotIn('environment',str(result))
 def test_other_live_private_process_refuses_before_open_or_signal(self):
  for other in (proc(11,comm='ArmaReforger',argv=(b'ArmaReforger.exe',)),proc(11,ppid=42),proc(11,prefix=m.PREFIX+b'/../pfx')):
   i=Inspector([proc(),other]);f=FakePidfds(i)
   with self.assertRaises(m.CleanupRefused):m._cleanup(i,f)
   self.assertEqual(f.opened,{});self.assertEqual(f.sent,[])
 def test_pid_reuse_after_open_refuses(self):
  i=Inspector([proc()]);f=FakePidfds(i)
  f.on_open=lambda pid:i.processes.update({pid:proc(pid,start=8888)})
  with self.assertRaises(m.CleanupRefused):m._cleanup(i,f)
  self.assertEqual(f.sent,[]);self.assertEqual(f.closed,[1010])
 def test_parent_reuse_or_role_change_refuses(self):
  for change in ('start','role','prefix','exe'):
   i=Inspector([proc()]);f=FakePidfds(i)
   def mutate(pid):
    if change=='start':i.parent=dataclasses.replace(PARENT,start=999)
    elif change=='role':i.processes[pid]=proc(pid,argv=(b'winedbg.exe',))
    elif change=='prefix':i.processes[pid]=proc(pid,prefix=b'/other/pfx')
    else:i.processes[pid]=proc(pid,executable_id=(9,9))
   f.on_open=mutate
   with self.subTest(change=change),self.assertRaises(m.CleanupRefused):m._cleanup(i,f)
   self.assertEqual(f.sent,[])
 def test_bounded_kill_uses_same_pinned_handles(self):
  i=Inspector([proc(10),proc(11)]);f=FakePidfds(i,term_exits=False)
  result=m._cleanup(i,f)
  self.assertEqual(f.sent,[(10,signal.SIGTERM),(11,signal.SIGTERM),(10,signal.SIGKILL),(11,signal.SIGKILL)])
  self.assertTrue(result['cleared']);self.assertEqual(result['kill_sent'],[10,11]);self.assertEqual(f.waits.count(3.0),1);self.assertEqual(f.waits.count(1.0),1)
 def test_new_active_process_after_partial_term_preserved(self):
  i=Inspector([proc(10),proc(11)]);f=FakePidfds(i)
  def new_process(pid,sig):
   if pid==10:i.processes[99]=proc(99,comm='active-game',argv=(b'game.exe',))
  f.on_send=new_process
  with self.assertRaises(m.CleanupRefused) as failure:m._cleanup(i,f)
  self.assertEqual(f.sent,[(10,signal.SIGTERM)]);self.assertEqual(failure.exception.report['term_sent'],[10]);self.assertFalse(failure.exception.report['cleared']);self.assertIn(11,i.processes);self.assertIn(99,i.processes)
 def test_partial_cleanup_failure_has_no_broad_fallback(self):
  i=Inspector([proc()]);f=FakePidfds(i,term_exits=False,kill_exits=False)
  with self.assertRaises(m.CleanupRefused) as failure:m._cleanup(i,f)
  self.assertEqual(failure.exception.report['kill_sent'],[10]);self.assertFalse(failure.exception.report['cleared']);self.assertEqual(f.closed,[1010]);self.assertEqual(len(f.sent),2)
 def test_new_adopted_helper_not_in_snapshot_refused(self):
  i=Inspector([proc()]);f=FakePidfds(i);f.on_open=lambda pid:i.processes.update({11:proc(11)})
  with self.assertRaises(m.CleanupRefused):m._cleanup(i,f)
  self.assertEqual(f.sent,[])
 def test_production_scope_rejects_other_prefix_before_backend(self):
  with patch.object(m,'Pidfds',side_effect=AssertionError('No signal backend permitted')):
   for other in ('/tmp/other','/home/user/.wine','/tmp/erri-headless-compat/../other'):
    with self.subTest(other=other),self.assertRaises(m.CleanupRefused):m.cleanup_private_prefix(other,'/installed/proton')
 def test_exited_zombies_are_not_signalled(self):
  i=Inspector([proc(state='Z')]);f=FakePidfds(i);r=m._cleanup(i,f);self.assertTrue(r['cleared']);self.assertEqual(f.opened,{});self.assertEqual(f.sent,[])
 def test_metadata_permissions_do_not_hide_runtime_candidates(self):
  self.assertTrue(m.native_metadata_proves_unrelated(PARENT))
  self.assertFalse(m.native_metadata_proves_unrelated(proc(exe=None,executable_id=None)))
  self.assertFalse(m.native_metadata_proves_unrelated(proc(comm='python',argv=(b'/usr/bin/python',b'/runtime/proton'),exe='/usr/bin/python')))
 def test_observed_nondumpable_native_services_are_unrelated(self):
  agent=proc(comm='ssh-agent',argv=(b'/usr/bin/ssh-agent',b'-D',b'-a',f'/run/user/{UID}/ssh-agent.socket'.encode()),exe=None,executable_id=None)
  self.assertTrue(m.native_metadata_proves_unrelated(agent))
  for comm,argv in [('kwin_wayland',(b'/usr/bin/kwin_wayland',)),('polkit-kde-auth',(b'/usr/lib/polkit-kde-authentication-agent-1',)),('apache2',(b'apache2',b'-DFOREGROUND'))]:
   self.assertTrue(m.native_metadata_proves_unrelated(dataclasses.replace(agent,comm=comm,argv=argv)))
  for mutation in ({'comm':'winedevice.exe'},{'argv':agent.argv+(b'proton',)},{'argv':(b'/usr/bin/something-else',)},{'argv':(b'ssh-agent',)},{'comm':'apache2','argv':(b'apache2',b'-unknown')},{'comm':'wine64-preloade','argv':(b'/runtime/wine64-preloader',)}):
   self.assertFalse(m.native_metadata_proves_unrelated(dataclasses.replace(agent,**mutation)))


 def test_unreadable_scdaemon_skipped_but_runtime_and_near_matches_refused(self):
  entry=MagicMock();entry.name='10';entry.stat.return_value.st_uid=UID
  entry.__truediv__.return_value.read_bytes.side_effect=PermissionError()
  root=MagicMock();root.iterdir.side_effect=lambda:iter([entry])
  inspector=m.ProcFS(EXE,INODE,root=root,uid=UID,self_pid=99)
  native=proc(comm='scdaemon',argv=(b'scdaemon',b'--multi-server'),exe=None,executable_id=None)
  with patch.object(inspector,'metadata',return_value=native):
   self.assertEqual(inspector.devices(),[])
  for candidate in [proc(exe=None,executable_id=None),
                    dataclasses.replace(native,comm='wine64-preloade'),
                    dataclasses.replace(native,argv=native.argv+(b'proton',)),
                    dataclasses.replace(native,argv=(b'scdaemon',b'--unknown'))]:
   with self.subTest(candidate=candidate),patch.object(inspector,'metadata',return_value=candidate),self.assertRaises(m.CleanupRefused):
    inspector.devices()

if __name__=='__main__':unittest.main()