PaperClipAI/tests/runner-e2e/pi-provider-fault.test.py

"""Credential-free admission tests; real pidfd calibration runs only on Linux."""
import importlib.util
import json
import os
from pathlib import Path
import shutil
import signal
import subprocess
import sys
import tempfile
import time
from types import SimpleNamespace
import unittest
import select
from unittest.mock import patch

spec = importlib.util.spec_from_file_location('pi_fault', Path(__file__).with_name('pi-provider-fault.py'))
fault = importlib.util.module_from_spec(spec)
spec.loader.exec_module(fault)
BOOT = '12345678-1234-1234-1234-123456789abc'
ROOT = {'pid': 21, 'ppid': 1, 'startTicks': '100', 'bootId': BOOT}
PARENT = {'pid': 22, 'ppid': 21, 'startTicks': '101', 'bootId': BOOT}
TARGET = {'pid': 23, 'ppid': 22, 'startTicks': '102', 'bootId': BOOT}
DIST = '/tmp/paperclip-acpx-native-fixture/distribution'
RUNTIME = '/workspace/.paperclip-runtime/paperclip-runner'
CANCELLED = False

def check_cancelled():
    if CANCELLED: raise KeyboardInterrupt('calibration cancelled')


NAMES = [fault.NODE, fault.ENTRY, fault.EXTENSION, 'pi-entry.cjs', 'node_modules/pi-acp/dist/index.js', 'node_modules/pi-acp/dist/paperclip-runtime.js']


class ClosureMetadataTests(unittest.TestCase):
    def setUp(self):
        self.entries = [
            {'path': name, 'sha256': fault.digest(name.encode()), 'size': len(name), 'executable': name == fault.NODE}
            for name in sorted(NAMES)
        ]
        self.canonical = json.dumps(self.entries, separators=(',', ':'), ensure_ascii=False).encode()
        self.pin = fault.digest(self.canonical)
        self.manifest = {'entries': self.entries}

    def test_canonical_entries_pin_admits_metadata(self):
        raw = json.dumps(self.manifest, indent=2).encode()
        self.assertNotEqual(fault.digest(raw), self.pin)
        self.assertEqual(fault.parse_closure_metadata(raw, self.pin), self.entries)

    def test_metadata_formatting_and_property_order_preserve_pin(self):
        reordered = {'entries': [dict(reversed(list(entry.items()))) for entry in self.entries]}
        for manifest in [self.manifest, reordered]:
            for indent in [None, 2, 4]:
                with self.subTest(indent=indent):
                    self.assertEqual(fault.parse_closure_metadata(json.dumps(manifest, indent=indent).encode(), self.pin), self.entries)

    def test_raw_metadata_hash_is_not_a_closure_pin(self):
        raw = json.dumps(self.manifest, indent=2).encode()
        with self.assertRaisesRegex(RuntimeError, 'closure_pin'):
            fault.parse_closure_metadata(raw, fault.digest(raw))

    def test_changed_entry_is_rejected_by_the_pinned_digest(self):
        changed = json.loads(json.dumps(self.manifest))
        changed['entries'][0]['sha256'] = '0' * 64
        with self.assertRaisesRegex(RuntimeError, 'closure_pin'):
            fault.parse_closure_metadata(json.dumps(changed).encode(), self.pin)

    def test_unsafe_duplicate_unsorted_and_invalid_metadata_fail_closed(self):
        mutations = [
            lambda m: m['entries'][0].update(path='../escape'),
            lambda m: m['entries'][0].update(path='.paperclip-native-entry.cjs'),
            lambda m: m['entries'].insert(1, m['entries'][0].copy()),
            lambda m: m['entries'].reverse(),
            lambda m: m['entries'][0].update(size=True),
            lambda m: m['entries'][0].update(executable=1),
            lambda m: m['entries'][0].update(extra='foreign'),
        ]
        for index, mutate in enumerate(mutations):
            changed = json.loads(json.dumps(self.manifest))
            mutate(changed)
            with self.subTest(index=index), self.assertRaises(RuntimeError):
                fault.parse_closure_metadata(json.dumps(changed).encode(), self.pin)
        for invalid in [None, [], {}, {'entries': []}]:
            with self.subTest(invalid=invalid), self.assertRaises(RuntimeError):
                fault.parse_closure_metadata(json.dumps(invalid).encode(), self.pin)


class IdentityTests(unittest.TestCase):
    def setUp(self):
        self.files = {DIST + '/' + n: (n.encode(), (1, 100 + i)) for i, n in enumerate(NAMES)}
        self.files[RUNTIME + '/bin/paperclip-runnerd'] = (b'runner', (1, 90))
        entries = [{'path': n, 'sha256': fault.digest(n.encode()), 'size': len(n), 'executable': n == fault.NODE} for n in sorted(NAMES)]
        closure = json.dumps({'entries': entries}, indent=2).encode()
        self.files[fault.PACK + '/provider-assets/pi/linux-x64/native-closure.json'] = (closure, (1, 80))
        self.config = {'root': ROOT, 'binding': {'remoteCwd': '/workspace', 'runId': 'run'}, 'runtimeEnvironmentLeaseId': 'workspace-id',
                       'runnerdSha256': 'sha256:' + fault.digest(b'runner'), 'closureSha256': fault.digest(json.dumps(entries, separators=(',', ':'), ensure_ascii=False).encode())}
        self.table = {21: ROOT, 22: PARENT, 23: TARGET}
        self.args = {21: [RUNTIME + '/bin/paperclip-runnerd', '--run-id', 'run', '--environment-lease-id', 'workspace-id', '--lifecycle-mode', 'per_turn', '--state-dir', RUNTIME + '/sessions/' + 'a' * 64 + '/runner'],
                     22: [DIST + '/' + fault.NODE, '--require', DIST + '/' + fault.GUARD, DIST + '/pi-entry.cjs'], 23: ['pi']}
        self.patches = [patch.object(fault, 'proc', side_effect=lambda pid, boot: self.table[pid]), patch.object(fault, 'argv', side_effect=lambda pid: self.args[pid]),
                        patch.object(fault, 'checked_runner_executable', side_effect=lambda p: self.files[p]),
                        patch.object(fault, 'checked_file', side_effect=lambda p, *a: self.files[p]), patch.object(Path, 'read_text', return_value=BOOT),
                        patch.object(os, 'listdir', return_value=['21', '22', '23']), patch.object(os, 'getpgid', return_value=21),
                        patch.object(os, 'lstat', return_value=SimpleNamespace(st_mode=0o40500)), patch.object(os.path, 'realpath', side_effect=lambda p: p),
                        patch.object(os, 'readlink', return_value='/workspace'), patch.object(os, 'stat', side_effect=lambda p: SimpleNamespace(st_dev=1, st_ino=90 if p == '/proc/21/exe' else 100))]
        for p in self.patches: p.start(); self.addCleanup(p.stop)

    def test_parent_attests_title_overwritten_child(self):
        receipt = fault.inspect(self.config)
        self.assertEqual(receipt['target'], TARGET)
        self.assertEqual(receipt['entrypointAttribution'], 'pinned_wrapper_parent')
        self.assertFalse(receipt['originalChildArgvAvailable'])
        self.assertEqual(receipt['ancestry'], [TARGET, PARENT, ROOT])

    def test_descriptor_launch_attests_the_same_pinned_wrapper(self):
        for descriptor in [3, 7]:
            self.args[22][0] = f'/proc/self/fd/{descriptor}'
            with self.subTest(descriptor=descriptor):
                self.assertEqual(fault.inspect(self.config)['target'], TARGET)

    def test_descriptor_launch_rejects_foreign_missing_or_unbound_fd(self):
        for name in ['/proc/self/fd/8', '/proc/99/fd/7', '/foreign/node']:
            self.args[22][0] = name
            with self.subTest(name=name), self.assertRaisesRegex(RuntimeError, 'wrapper_parent'):
                fault.inspect(self.config)
        self.args[22][0] = '/proc/self/fd/7'
        stat_original = os.stat
        for problem in ['foreign', 'missing']:
            def descriptor_stat(path):
                if path == '/proc/22/fd/7':
                    if problem == 'missing': raise FileNotFoundError(path)
                    return SimpleNamespace(st_dev=1, st_ino=999)
                return stat_original(path)
            with self.subTest(problem=problem), patch.object(os, 'stat', side_effect=descriptor_stat), self.assertRaises((RuntimeError, FileNotFoundError)):
                fault.inspect(self.config)

    def test_wrong_metadata_never_selects(self):
        for field, value in [('closureSha256', '0' * 64), ('runtimeEnvironmentLeaseId', 'foreign'), ('runnerdSha256', 'sha256:' + '0' * 64)]:
            with self.subTest(field=field), self.assertRaises(RuntimeError): fault.inspect({**self.config, field: value})

    def test_wrong_title_is_not_a_generic_node_selector(self):
        self.args[23] = ['node']
        with self.assertRaisesRegex(RuntimeError, 'pi_process_title'): fault.inspect(self.config)

    def test_foreign_parent_and_changed_wrapper_are_rejected(self):
        self.args[22][3] = '/foreign/pi-entry.cjs'
        with self.assertRaisesRegex(RuntimeError, 'unique_pi_child'): fault.inspect(self.config)
        self.args[22][3] = DIST + '/pi-entry.cjs'
        self.files[DIST + '/node_modules/pi-acp/dist/index.js'] = (b'changed', (1, 104))
        with self.assertRaisesRegex(RuntimeError, 'snapshot_digest'): fault.inspect(self.config)

    def test_foreign_root_reused_pid_and_wrong_cwd_are_rejected(self):
        for mutation in [{'startTicks': '999'}, {'bootId': 'foreign'}, {'ppid': 99}]:
            with self.subTest(mutation=mutation), self.assertRaises(RuntimeError): fault.inspect({**self.config, 'root': {**ROOT, **mutation}})
        with patch.object(os, 'readlink', return_value='/foreign'), self.assertRaisesRegex(RuntimeError, 'cwd'): fault.inspect(self.config)

    def test_ambiguous_direct_children_are_rejected(self):
        self.table[24] = {**TARGET, 'pid': 24}; self.args[24] = ['pi']
        with patch.object(os, 'listdir', return_value=['21', '22', '23', '24']), self.assertRaisesRegex(RuntimeError, 'unique_pi_child'): fault.inspect(self.config)


class RunnerExecutableTests(unittest.TestCase):
    def test_regular_and_preinstalled_link_resolve_to_the_same_inode(self):
        with tempfile.TemporaryDirectory() as tmp:
            executable = Path(tmp).resolve() / 'installed-runner'
            executable.write_bytes(b'pinned runner')
            link = executable.with_name('runtime-runner')
            link.symlink_to(executable)
            content, inode = fault.checked_runner_executable(str(link))
            self.assertEqual(content, b'pinned runner')
            self.assertEqual((content, inode), fault.checked_runner_executable(str(executable)))

    def test_retargeted_link_is_rejected_after_read(self):
        with tempfile.TemporaryDirectory() as tmp:
            executable = Path(tmp).resolve() / 'installed-runner'
            executable.write_bytes(b'pinned runner')
            foreign = executable.with_name('foreign-runner')
            foreign.write_bytes(b'foreign')
            link = executable.with_name('runtime-runner')
            link.symlink_to(executable)
            original = fault.checked_file
            def replace_during_read(*args):
                result = original(*args)
                link.unlink()
                link.symlink_to(foreign)
                return result
            with patch.object(fault, 'checked_file', side_effect=replace_during_read), self.assertRaisesRegex(RuntimeError, 'runner_link_changed'):
                fault.checked_runner_executable(str(link))

    def test_missing_link_and_directory_never_admit_an_executable(self):
        with tempfile.TemporaryDirectory() as tmp:
            root = Path(tmp).resolve()
            link = root / 'runtime-runner'
            link.symlink_to(root / 'missing')
            with self.assertRaises(FileNotFoundError): fault.checked_runner_executable(str(link))
            with self.assertRaisesRegex(RuntimeError, 'runner_file_shape'): fault.checked_runner_executable(str(root))


class PidfdTests(unittest.TestCase):
    def test_changed_or_retired_pidfd_never_signals_and_always_closes(self):
        before = {'target': TARGET}
        for second, fdinfo in [({'target': {**TARGET, 'startTicks': '999'}}, 'Pid:\t23\n'), (before, 'Pid:\t-1\n')]:
            with patch.object(sys, 'platform', 'linux'), patch.object(fault, 'inspect', side_effect=[before, second]), \
                 patch.object(os, 'pidfd_open', return_value=42, create=True), patch.object(signal, 'pidfd_send_signal', create=True) as sent, \
                 patch.object(os, 'close') as closed, patch.object(Path, 'read_text', return_value=fdinfo):
                with self.assertRaises(RuntimeError): fault.terminate({'root': ROOT})
                sent.assert_not_called(); closed.assert_called_once_with(42)

    def test_symlink_and_mutable_snapshot_file_fail(self):
        with tempfile.TemporaryDirectory() as tmp:
            path = Path(tmp).resolve() / 'data'; path.write_text('data')
            link = Path(tmp) / 'link'; link.symlink_to(path)
            with self.assertRaises(RuntimeError): fault.checked_file(str(link), 100, True)
            with self.assertRaises(RuntimeError): fault.checked_file(str(path), 100, True)
            path.chmod(0o400)
            self.assertEqual(fault.checked_file(str(path), 100, True)[0], b'data')

    @unittest.skipUnless(sys.platform == 'linux' and hasattr(os, 'pidfd_open'), 'Hosted native Linux calibration required')
    def test_real_title_overwrite_then_exact_child_pidfd(self):
        self.calibrate_real_child()

    @unittest.skipUnless(sys.platform == 'linux' and hasattr(os, 'pidfd_open'), 'Hosted native Linux descriptor calibration required')
    def test_real_descriptor_launch_then_exact_child_pidfd(self):
        for descriptor in [3, 7]:
            with self.subTest(descriptor=descriptor): self.calibrate_real_child(descriptor)

    @unittest.skipUnless(sys.platform == 'linux' and hasattr(os, 'pidfd_open'), 'Hosted native Linux preinstalled-link calibration required')
    def test_real_preinstalled_runner_link_then_exact_child_pidfd(self):
        self.calibrate_real_child(7, runner_link=True)

    def calibrate_real_child(self, descriptor=None, runner_link=False):
        node = shutil.which('node'); self.assertIsNotNone(node)
        base = Path(tempfile.mkdtemp(prefix='pi-fault-calibration-', dir='/tmp'))
        snapshot = Path(tempfile.mkdtemp(prefix='paperclip-acpx-native-', dir='/tmp'))
        distribution = snapshot / 'distribution'; workspace = base / 'workspace'; workspace.mkdir()
        runtime = workspace / '.paperclip-runtime/paperclip-runner'; (runtime / 'bin').mkdir(parents=True)
        pack = base / 'pack'; (pack / 'provider-assets/pi/linux-x64').mkdir(parents=True)
        root_child = None
        owned_fds = []
        try:
            for name in NAMES + [fault.GUARD]:
                p = distribution / name; p.parent.mkdir(parents=True, exist_ok=True); p.write_text('// fixture\n')
            shutil.copyfile(node, distribution / fault.NODE); (distribution / fault.NODE).chmod(0o500)
            runner = runtime / 'bin/paperclip-runnerd'
            if runner_link:
                installed = base / 'preinstalled-runner'
                shutil.copyfile(node, installed); installed.chmod(0o500)
                runner.symlink_to(installed)
            else:
                shutil.copyfile(node, runner); runner.chmod(0o500)
            ready = workspace / 'child.json'; exited = workspace / 'exit.json'
            (distribution / fault.ENTRY).write_text('process.title="pi"; require("node:fs").writeFileSync(' + json.dumps(str(ready)) + ',JSON.stringify({pid:process.pid}));setInterval(()=>{},1000);')
            (distribution / 'pi-entry.cjs').write_text('const p=require("node:child_process").spawn(process.execPath,["--require",' + json.dumps(str(distribution / fault.GUARD)) + ',' + json.dumps(str(distribution / fault.ENTRY)) + '],{stdio:"ignore"});p.on("exit",(code,signal)=>require("node:fs").writeFileSync(' + json.dumps(str(exited)) + ',JSON.stringify({code,signal})));process.on("SIGTERM",()=>{if(p.exitCode!==null||p.signalCode!==null)process.exit(0);p.once("exit",()=>process.exit(0));p.kill("SIGTERM")});setInterval(()=>{},1000);')
            entries = []
            for name in sorted(NAMES):
                p = distribution / name; content = p.read_bytes(); entries.append({'path': name, 'sha256': fault.digest(content), 'size': len(content), 'executable': name == fault.NODE})
            closure = json.dumps({'entries': entries}).encode(); (pack / 'provider-assets/pi/linux-x64/native-closure.json').write_bytes(closure)
            for directory, dirs, files in os.walk(snapshot):
                for f in files:
                    p = Path(directory) / f; p.chmod(0o500 if p == distribution / fault.NODE else 0o400)
                Path(directory).chmod(0o500)
            launch = json.dumps(str(distribution / fault.NODE))
            setup, stdio, release = '', '"ignore"', ''
            if descriptor is not None:
                # Production nativeBootstrap keeps its executable descriptor
                # at child FD 3, or FD 7 when lifetime/credential fences exist.
                setup = 'const fd=require("node:fs").openSync(' + launch + ',"r");'
                launch = json.dumps(f'/proc/self/fd/{descriptor}')
                stdio = '[' + ','.join(['"ignore"'] * descriptor + ['fd']) + ']'
                release = 'require("node:fs").closeSync(fd);'
            script = base / 'root.cjs'; script.write_text(setup + 'const p=require("node:child_process").spawn(' + launch + ',["--require",' + json.dumps(str(distribution / fault.GUARD)) + ',' + json.dumps(str(distribution / 'pi-entry.cjs')) + '],{stdio:' + stdio + '});' + release + 'process.on("SIGTERM",()=>{if(p.exitCode!==null||p.signalCode!==null)process.exit(0);p.once("exit",()=>process.exit(0));p.kill("SIGTERM")});setInterval(()=>{},1000);')
            check_cancelled()
            root_child = subprocess.Popen([str(runtime / 'bin/paperclip-runnerd'), str(script), '--run-id', 'fixture', '--environment-lease-id', 'workspace-id', '--lifecycle-mode', 'per_turn', '--state-dir', str(runtime / 'sessions' / ('a' * 64) / 'runner')], cwd=workspace, env={'PATH': '/usr/bin:/bin'}, start_new_session=True, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
            try: owned_fds.append(os.pidfd_open(root_child.pid))
            except Exception:
                # Direct Popen child has not been polled/reaped, so its numeric
                # identity cannot have been reused before this emergency stop.
                root_child.terminate(); root_child.wait(timeout=3)
                raise
            check_cancelled()
            deadline = time.monotonic() + 5
            while not ready.exists() and time.monotonic() < deadline:
                check_cancelled(); time.sleep(.02)
            self.assertTrue(ready.exists()); target = json.loads(ready.read_text())['pid']
            self.assertEqual(fault.argv(target), ['pi'], 'real Node process.title must overwrite argv')
            boot = Path('/proc/sys/kernel/random/boot_id').read_text().strip()
            config = {'root': fault.proc(root_child.pid, boot), 'binding': {'remoteCwd': str(workspace), 'runId': 'fixture'}, 'runtimeEnvironmentLeaseId': 'workspace-id', 'runnerdSha256': 'sha256:' + fault.digest((runtime / 'bin/paperclip-runnerd').read_bytes()), 'closureSha256': fault.digest(json.dumps(entries, separators=(',', ':'), ensure_ascii=False).encode())}
            with patch.object(fault, 'PACK', str(pack)):
                admitted = fault.inspect(config)
                self.assertEqual(admitted['target']['pid'], target)
                for identity in admitted['ancestry'][:-1]:
                    fd = os.pidfd_open(identity['pid'])
                    try:
                        fields = dict(line.split(':', 1) for line in Path(f'/proc/self/fdinfo/{fd}').read_text().splitlines() if ':' in line)
                        self.assertEqual(int(fields.get('Pid', '-1').strip()), identity['pid'])
                        self.assertEqual(fault.proc(identity['pid'], boot), identity)
                        self.assertEqual(fault.inspect(config), admitted)
                    except BaseException:
                        os.close(fd); raise
                    owned_fds.append(fd)
                    check_cancelled()
                check_cancelled()
                receipt = fault.terminate(config)
            self.assertEqual(receipt['target']['pid'], target); self.assertTrue(receipt['signalled']); self.assertEqual(select.select([owned_fds[0]], [], [], 0)[0], [])
            deadline = time.monotonic() + 3
            while not exited.exists() and time.monotonic() < deadline:
                check_cancelled(); time.sleep(.02)
            self.assertEqual(json.loads(exited.read_text()), {'code': None, 'signal': 'SIGKILL'})
        finally:
            try:
                if root_child is not None and owned_fds:
                    # A pidfd never retargets a recycled numeric process group.
                    # Before readiness the synthetic root owns a SIGTERM cascade.
                    try: signal.pidfd_send_signal(owned_fds[0], signal.SIGTERM, None, 0)
                    except ProcessLookupError: pass
                    try: root_child.wait(timeout=3)
                    except subprocess.TimeoutExpired:
                        for fd in reversed(owned_fds):
                            try: signal.pidfd_send_signal(fd, signal.SIGKILL, None, 0)
                            except ProcessLookupError: pass
                        root_child.wait(timeout=3)
                    # Also retire individually pinned descendants if root exited
                    # unexpectedly. Readable pidfds identify already-dead children.
                    for fd in reversed(owned_fds[1:]):
                        if not select.select([fd], [], [], 0)[0]:
                            try: signal.pidfd_send_signal(fd, signal.SIGKILL, None, 0)
                            except ProcessLookupError: pass
                        self.assertTrue(select.select([fd], [], [], 3)[0], 'owned descendant must retire')
            finally:
                for fd in owned_fds: os.close(fd)
                for directory, dirs, files in os.walk(snapshot): os.chmod(directory, 0o700)
                shutil.rmtree(snapshot); shutil.rmtree(base)


if __name__ == '__main__':
    def cancelled(_signum, _frame):
        # Flag only: never interrupt Popen or pidfd ownership publication.
        global CANCELLED
        CANCELLED = True
    signal.signal(signal.SIGTERM, cancelled)
    signal.signal(signal.SIGINT, cancelled)
    unittest.main()