245 lines
13 KiB
Python
245 lines
13 KiB
Python
"""Descriptor/pidfd mechanisms; full cgroup proof is the real PID1 fixture."""
|
|
import os
|
|
from pathlib import Path
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
import unittest
|
|
import uuid
|
|
from unittest.mock import patch
|
|
sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
|
|
from c01_access_kernel import Cohort
|
|
import c01_access_kernel as kernel
|
|
|
|
|
|
class RetirementPins(unittest.TestCase):
|
|
def child(self):
|
|
child=subprocess.Popen(['/usr/bin/sleep','30'])
|
|
def cleanup():
|
|
if child.poll() is None:child.kill();child.wait()
|
|
self.addCleanup(cleanup)
|
|
return child
|
|
|
|
def test_only_exact_pinned_task_missing_read_can_retire(self):
|
|
child=self.child()
|
|
with kernel.PinnedTask(child.pid) as task:
|
|
item=task.record()
|
|
def exited():
|
|
child.terminate();child.wait(timeout=5)
|
|
return task.namespace('net')
|
|
result=kernel.inspect_task(task,item,exited)
|
|
self.assertEqual(result,{'pid':child.pid,'start':item['start'],'retired_pidfd':True})
|
|
|
|
def test_missing_appointed_namespace_is_fatal_even_after_child_exit(self):
|
|
child=self.child();missing=Path('/run')/('c01-no-such-namespace-'+uuid.uuid4().hex)
|
|
with kernel.PinnedTask(child.pid) as task:
|
|
item=task.record()
|
|
def failed():
|
|
child.terminate();child.wait(timeout=5)
|
|
# Actual independent appointment validation, not a mock error.
|
|
kernel.c01_network.require_private(child.pid,'managed')
|
|
with patch.dict(kernel.c01_network.NAMES,{'managed':missing}):
|
|
with self.assertRaises(FileNotFoundError):kernel.inspect_task(task,item,failed)
|
|
|
|
def test_standalone_namespace_check_does_not_hide_missing_appointment(self):
|
|
child=self.child()
|
|
with kernel.PinnedTask(child.pid) as task:
|
|
item=task.record()
|
|
def failed():
|
|
child.terminate();child.wait(timeout=5)
|
|
kernel.require_task_network(task,'managed')
|
|
with patch.dict(kernel.c01_network.NAMES,{'managed':Path('/run')/uuid.uuid4().hex}):
|
|
with self.assertRaises(FileNotFoundError):kernel.inspect_task(task,item,failed)
|
|
|
|
def test_independent_executable_and_view_errors_are_never_retirement(self):
|
|
for source in ('executable','view'):
|
|
with self.subTest(source=source):
|
|
child=self.child()
|
|
with kernel.PinnedTask(child.pid) as task:
|
|
item=task.record();root=task.root()
|
|
try:
|
|
def failed():
|
|
child.terminate();child.wait(timeout=5)
|
|
if source=='executable':kernel.protected(Path('/run')/uuid.uuid4().hex)
|
|
else:kernel.view_stat(root,'/run/'+uuid.uuid4().hex)
|
|
with self.assertRaises(FileNotFoundError):kernel.inspect_task(task,item,failed)
|
|
finally:os.close(root)
|
|
|
|
def test_independent_parent_exit_cannot_be_excused_by_child_exit(self):
|
|
child=self.child();parent=self.child()
|
|
with kernel.PinnedTask(child.pid) as task, kernel.PinnedTask(parent.pid) as parent_task:
|
|
item=task.record()
|
|
def failed():
|
|
child.terminate();child.wait(timeout=5)
|
|
parent.terminate();parent.wait(timeout=5)
|
|
parent_task.namespace('net')
|
|
with self.assertRaisesRegex(ValueError,'independent task lifetime lost'):
|
|
kernel.inspect_task(task,item,failed)
|
|
|
|
def test_live_unreadable_or_missing_task_fields_are_fatal(self):
|
|
child=self.child()
|
|
with kernel.PinnedTask(child.pid) as task:
|
|
item=task.record();real_open=os.open
|
|
for exception in (FileNotFoundError('ns/net'),PermissionError('ns/net')):
|
|
def unavailable(path,*args,**kw):
|
|
if path=='ns/net' and kw.get('dir_fd')==task.procfd:raise exception
|
|
return real_open(path,*args,**kw)
|
|
with patch.object(kernel.os,'open',side_effect=unavailable):
|
|
with self.assertRaises((ValueError,PermissionError)):
|
|
kernel.inspect_task(task,item,lambda:task.namespace('net'))
|
|
self.assertIsNone(child.poll())
|
|
|
|
def test_exit_teardown_wait_requires_real_pidfd_completion(self):
|
|
child=self.child()
|
|
with kernel.PinnedTask(child.pid) as task:
|
|
item=task.record();real_open=os.open
|
|
def teardown(path,*args,**kw):
|
|
if path=='ns/net' and kw.get('dir_fd')==task.procfd:
|
|
child.terminate() # deliberately do not reap/wait before read
|
|
raise FileNotFoundError('nsproxy teardown')
|
|
return real_open(path,*args,**kw)
|
|
with patch.object(kernel.os,'open',side_effect=teardown):
|
|
result=kernel.inspect_task(task,item,lambda:task.namespace('net'))
|
|
self.assertTrue(result['retired_pidfd']);self.assertTrue(task.exited())
|
|
child.wait(timeout=5)
|
|
|
|
def test_later_pid_reuse_seam_cannot_borrow_successor_proc_evidence(self):
|
|
child=self.child();successor=self.child()
|
|
with kernel.PinnedTask(child.pid) as task:
|
|
item=task.record();real_open=os.open;numeric_reads=[]
|
|
def reused(path,*args,**kw):
|
|
text=os.fspath(path)
|
|
if text.startswith(f'/proc/{child.pid}/'):
|
|
numeric_reads.append(text)
|
|
path=text.replace(f'/proc/{child.pid}/',f'/proc/{successor.pid}/',1)
|
|
return real_open(path,*args,**kw)
|
|
def exit_then_read():
|
|
child.terminate();child.wait(timeout=5)
|
|
# Redirect any late numeric reopening to a live successor. The
|
|
# retained directory must instead remain the original task's.
|
|
with patch.object(kernel.os,'open',side_effect=reused):return task.namespace('net')
|
|
result=kernel.inspect_task(task,item,exit_then_read)
|
|
self.assertTrue(result['retired_pidfd']);self.assertEqual(numeric_reads,[])
|
|
self.assertNotIn('namespaces',result);self.assertNotIn('uids',result)
|
|
self.assertIsNone(successor.poll())
|
|
|
|
def test_pid_reuse_during_initial_proc_binding_is_rejected(self):
|
|
child=self.child();successor=self.child();real_open=os.open
|
|
def reused(path,*args,**kw):
|
|
if path==str(child.pid) and kw.get('dir_fd') is not None:
|
|
child.terminate();child.wait(timeout=5)
|
|
return real_open(str(successor.pid),*args,**kw)
|
|
return real_open(path,*args,**kw)
|
|
with patch.object(kernel.os,'open',side_effect=reused):
|
|
with self.assertRaises(ValueError):kernel.PinnedTask(child.pid)
|
|
self.assertIsNone(successor.poll())
|
|
|
|
def test_foreign_proc_directory_cannot_bind_to_live_original_pidfd(self):
|
|
child=self.child();successor=self.child();real_open=os.open
|
|
def foreign(path,*args,**kw):
|
|
if path==str(child.pid) and kw.get('dir_fd') is not None:
|
|
return real_open(str(successor.pid),*args,**kw)
|
|
return real_open(path,*args,**kw)
|
|
with patch.object(kernel.os,'open',side_effect=foreign):
|
|
with self.assertRaisesRegex(ValueError,'proc directory PID differs'):
|
|
kernel.PinnedTask(child.pid)
|
|
self.assertIsNone(child.poll());self.assertIsNone(successor.poll())
|
|
|
|
def test_invalid_namespace_capability_is_fatal_even_after_task_exit(self):
|
|
child=self.child()
|
|
with kernel.PinnedTask(child.pid) as task:
|
|
item=task.record();real_open=os.open
|
|
def wrong_capability(path,*args,**kw):
|
|
if path=='ns/net' and kw.get('dir_fd')==task.procfd:
|
|
return real_open('/dev/null',os.O_RDONLY|os.O_CLOEXEC)
|
|
return real_open(path,*args,**kw)
|
|
def failed():
|
|
child.terminate();child.wait(timeout=5)
|
|
with patch.object(kernel.os,'open',side_effect=wrong_capability):return task.namespace('net')
|
|
with self.assertRaises(OSError):kernel.inspect_task(task,item,failed)
|
|
|
|
def test_after_inspection_validation_discards_evidence_of_exited_task(self):
|
|
child=self.child()
|
|
with kernel.PinnedTask(child.pid) as task:
|
|
item=task.record()
|
|
def finish_then_exit():
|
|
evidence={'namespaces':dict(item['namespaces']),'group':item['group']}
|
|
child.terminate();child.wait(timeout=5)
|
|
return evidence
|
|
result=kernel.inspect_task(task,item,finish_then_exit)
|
|
self.assertTrue(result['retired_pidfd']);self.assertNotIn('namespaces',result)
|
|
|
|
def test_live_changed_credentials_or_namespace_evidence_refuses(self):
|
|
child=self.child()
|
|
with kernel.PinnedTask(child.pid) as task:
|
|
item=task.record()
|
|
for field,changed in (('uids',[12345]*4),('group','/sys/fs/cgroup/foreign'),('namespaces',{})):
|
|
with patch.object(task,'record',return_value={**item,field:changed}):
|
|
with self.assertRaisesRegex(ValueError,'incarnation changed'):
|
|
kernel.inspect_task(task,item,lambda:{'admitted':True})
|
|
|
|
def test_changed_sample_is_discarded_and_all_predicates_recheck_fresh_record(self):
|
|
child=self.child()
|
|
with kernel.PinnedTask(child.pid) as task:
|
|
current=task.record();old={**current,'ppid':current['ppid']+1};seen=[]
|
|
def inspect(item):
|
|
seen.append(item.copy())
|
|
return {'checked_ppid':item['ppid']}
|
|
final,evidence=kernel.stable_task_evidence(task,old,inspect)
|
|
self.assertEqual(seen,[old,current]);self.assertEqual(final,current)
|
|
self.assertEqual(evidence,{'checked_ppid':current['ppid']})
|
|
|
|
def test_unstable_task_sampling_is_bounded_and_parent_changes_are_fatal(self):
|
|
child=self.child();parent=self.child()
|
|
with kernel.PinnedTask(child.pid) as task, kernel.PinnedTask(parent.pid) as parent_task:
|
|
item=task.record();calls=[]
|
|
def inspect(_):calls.append(1);return {'discarded':True}
|
|
with patch.object(task,'finish',side_effect=kernel.TaskEvidenceChanged(task,['group'])):
|
|
with self.assertRaises(kernel.TaskEvidenceChanged):kernel.stable_task_evidence(task,item,inspect)
|
|
self.assertEqual(len(calls),2)
|
|
def parent_failure(_):
|
|
child.terminate();child.wait(timeout=5)
|
|
raise kernel.TaskEvidenceChanged(parent_task,['namespaces'])
|
|
with self.assertRaises(kernel.TaskEvidenceChanged):kernel.stable_task_evidence(task,item,parent_failure)
|
|
|
|
def test_fresh_sample_cannot_hide_policy_failure(self):
|
|
child=self.child()
|
|
with kernel.PinnedTask(child.pid) as task:
|
|
current=task.record();old={**current,'ppid':current['ppid']+1};calls=[]
|
|
def inspect(item):
|
|
calls.append(1)
|
|
if item==current:raise PermissionError('fresh policy check refused')
|
|
return {'stale':True}
|
|
with self.assertRaises(PermissionError):kernel.stable_task_evidence(task,old,inspect)
|
|
self.assertEqual(len(calls),2)
|
|
|
|
def test_independent_revalidation_also_runs_on_task_retirement(self):
|
|
child=self.child()
|
|
with tempfile.TemporaryDirectory() as directory, kernel.PinnedTask(child.pid) as task:
|
|
appointment=Path(directory)/'appointment';appointment.write_text('retained')
|
|
task.dependencies.append(lambda:appointment.stat());item=task.record()
|
|
def failed():
|
|
appointment.unlink();child.terminate();child.wait(timeout=5)
|
|
task.namespace('net')
|
|
with self.assertRaises(FileNotFoundError):kernel.inspect_task(task,item,failed)
|
|
|
|
def test_unlinked_directory_does_not_retire_a_live_captured_process(self):
|
|
with tempfile.TemporaryDirectory() as temporary:
|
|
path=Path(temporary)/'cohort';path.mkdir();child=self.child()
|
|
group=Cohort.__new__(Cohort);group.path=path
|
|
group.fd=os.open(path,os.O_PATH|os.O_DIRECTORY);pidfd=os.pidfd_open(child.pid)
|
|
group.pidfds={child.pid:pidfd};group.spec={'members':{str(child.pid):'mechanism-test'}}
|
|
try:
|
|
path.rmdir();self.assertTrue(group.unlinked());self.assertFalse(group.retired())
|
|
child.terminate();child.wait(timeout=5);self.assertTrue(group.retired())
|
|
group.pidfds={};self.assertFalse(group.retired(),'missing captured kernel references must refuse')
|
|
finally:os.close(pidfd);os.close(group.fd)
|
|
|
|
def test_missing_locator_is_not_deletion_of_the_retained_object(self):
|
|
with tempfile.TemporaryDirectory() as temporary:
|
|
path=Path(temporary)/'cohort';path.mkdir()
|
|
group=Cohort.__new__(Cohort);group.path=path;group.fd=os.open(path,os.O_PATH|os.O_DIRECTORY)
|
|
try:
|
|
path.rename(Path(temporary)/'elsewhere');self.assertFalse(path.exists());self.assertFalse(group.unlinked())
|
|
finally:os.close(group.fd)
|