130 lines
8.8 KiB
Python
130 lines
8.8 KiB
Python
"""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
|
|
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_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)))
|
|
|
|
if __name__=='__main__':unittest.main()
|