69 lines
3.3 KiB
Python
69 lines
3.3 KiB
Python
"""Real exec/pipe tests for trusted Runner program framing, without Docker."""
|
||
from pathlib import Path
|
||
import os
|
||
import subprocess
|
||
import sys
|
||
import tempfile
|
||
import unittest
|
||
|
||
SOURCE = Path(__file__).resolve().parents[1] / 'crates/supervisor/src/runner_bootstrap.py'
|
||
|
||
|
||
class BootstrapTests(unittest.TestCase):
|
||
def setUp(self):
|
||
if os.geteuid() != 0:
|
||
self.skipTest('protected Runner bootstrap requires root')
|
||
self.temporary = tempfile.TemporaryDirectory()
|
||
self.addCleanup(self.temporary.cleanup)
|
||
self.root = Path(self.temporary.name)
|
||
self.bootstrap = SOURCE.read_text().replace("root = '/var/lib/lazyboy-runner'", f'root = {str(self.root)!r}')
|
||
|
||
def test_program_larger_than_single_argv_limit_preserves_request_bytes(self):
|
||
program = ('# ' + '程式' * 40000 + '\nimport sys\nsys.stdout.buffer.write(sys.stdin.buffer.read())\n').encode()
|
||
self.assertGreater(len(program), 128 * 1024)
|
||
request = '{"input":"PRIVATE_REQUEST_CANARY：原始資料\\n第二行"}'.encode()
|
||
frame = f'LAZYBOY_RUNNER_V1 {len(program)}\n'.encode() + program + request
|
||
child = subprocess.Popen([sys.executable, '-I', '-c', self.bootstrap], stdin=subprocess.PIPE,
|
||
stdout=subprocess.PIPE, stderr=subprocess.PIPE)
|
||
try:
|
||
for offset in range(0, len(frame), 4093):
|
||
child.stdin.write(frame[offset:offset + 4093])
|
||
child.stdin.flush()
|
||
child.stdin.close()
|
||
self.assertEqual(child.stdout.read(), request)
|
||
self.assertEqual(child.wait(timeout=5), 0, child.stderr.read())
|
||
finally:
|
||
if child.poll() is None:
|
||
child.kill()
|
||
child.wait()
|
||
child.stdout.close()
|
||
child.stderr.close()
|
||
installed = self.root / 'runner_jobs.py'
|
||
self.assertEqual(installed.read_bytes(), program)
|
||
self.assertNotIn(b'PRIVATE_REQUEST_CANARY', installed.read_bytes())
|
||
self.assertEqual(installed.stat().st_mode & 0o777, 0o600)
|
||
self.assertEqual(list(self.root.iterdir()), [installed])
|
||
|
||
def test_invalid_or_truncated_frame_never_replaces_the_program(self):
|
||
installed = self.root / 'runner_jobs.py'
|
||
installed.write_bytes(b'previous program')
|
||
for frame in [b'', b'wrong 2\nxx', b'LAZYBOY_RUNNER_V1 -2\nxx', b'LAZYBOY_RUNNER_V1 0\n',
|
||
b'LAZYBOY_RUNNER_V1 1048577\n', b'x' * 64, b'LAZYBOY_RUNNER_V1 99\nshort']:
|
||
result = subprocess.run([sys.executable, '-I', '-c', self.bootstrap], input=frame,
|
||
capture_output=True, timeout=5)
|
||
self.assertNotEqual(result.returncode, 0, frame)
|
||
self.assertEqual(installed.read_bytes(), b'previous program')
|
||
self.assertEqual(list(self.root.iterdir()), [installed])
|
||
|
||
def test_unprotected_program_directory_is_refused(self):
|
||
self.root.chmod(0o777)
|
||
result = subprocess.run([sys.executable, '-I', '-c', self.bootstrap], input=b'LAZYBOY_RUNNER_V1 4\npass',
|
||
capture_output=True, timeout=5)
|
||
self.assertNotEqual(result.returncode, 0)
|
||
self.assertIn(b'unprotected Runner directory', result.stderr)
|
||
self.assertFalse((self.root / 'runner_jobs.py').exists())
|
||
|
||
|
||
if __name__ == '__main__':
|
||
unittest.main()
|