232 lines
13 KiB
Python
232 lines
13 KiB
Python
|
"""Run INSIDE a disposable Computer as root; every job executes as uid 1000."""
|
||
|
|
import base64
|
||
|
|
import json
|
||
|
|
import os
|
||
|
|
from pathlib import Path
|
||
|
|
import subprocess
|
||
|
|
import sqlite3
|
||
|
|
import signal
|
||
|
|
import time
|
||
|
|
import unittest
|
||
|
|
import uuid
|
||
|
|
|
||
|
|
RUNNER = '/var/lib/lazyboy-runner/runner_jobs.py'
|
||
|
|
|
||
|
|
|
||
|
|
class RunnerJobs(unittest.TestCase):
|
||
|
|
def setUp(self):
|
||
|
|
if os.environ.get('LAZYBOY_RUNNER_TEST_CONTAINER') != '1' or not Path('/.dockerenv').exists():
|
||
|
|
self.fail('run only inside the disposable Runner test container')
|
||
|
|
self.jobs = []
|
||
|
|
self.identity = {'computer_id': 'fixture', 'bot_id': 'bot-a', 'generation': 1}
|
||
|
|
|
||
|
|
def call(self, request, check=True, user=None):
|
||
|
|
prefix = ['runuser', '-u', user, '--'] if user else []
|
||
|
|
result = subprocess.run(prefix + ['python3', '-I', RUNNER], input=json.dumps(request),
|
||
|
|
capture_output=True, text=True, timeout=10)
|
||
|
|
if check:
|
||
|
|
self.assertEqual(result.returncode, 0, result.stdout + result.stderr)
|
||
|
|
return json.loads(result.stdout)
|
||
|
|
|
||
|
|
def start(self, argv, **kwargs):
|
||
|
|
job = 'test-' + uuid.uuid4().hex
|
||
|
|
self.jobs.append(job)
|
||
|
|
request = dict(action='start', job_id=job, identity=self.identity, argv=argv,
|
||
|
|
runtime_timeout_ms=5000, **kwargs)
|
||
|
|
return request, self.call(request)
|
||
|
|
|
||
|
|
def follow(self, job, action='status', **kwargs):
|
||
|
|
return self.call(dict(action=action, job_id=job, identity=self.identity, **kwargs))
|
||
|
|
|
||
|
|
def wait(self, job):
|
||
|
|
deadline = time.monotonic() + 8
|
||
|
|
while time.monotonic() < deadline:
|
||
|
|
result = self.follow(job)
|
||
|
|
if result['status'] not in ('accepted', 'running', 'paused', 'cancelling'):
|
||
|
|
return result
|
||
|
|
time.sleep(.03)
|
||
|
|
self.fail('job did not finish')
|
||
|
|
|
||
|
|
def tearDown(self):
|
||
|
|
for job in self.jobs:
|
||
|
|
self.follow(job, 'cancel')
|
||
|
|
self.wait(job)
|
||
|
|
|
||
|
|
def test_durable_identity_replay_output_cursor_and_binary(self):
|
||
|
|
request, first = self.start(['python3', '-c', "import sys,time;sys.stdout.buffer.write(b'\\xffA');sys.stdout.flush();time.sleep(.2);sys.stdout.write('B');sys.stderr.write('error');sys.exit(7)"])
|
||
|
|
self.assertEqual(self.call(request)['job_id'], first['job_id'])
|
||
|
|
changed = dict(request, argv=['true'])
|
||
|
|
self.assertEqual(self.call(changed, check=False)['error'], 'PAYLOAD_MISMATCH')
|
||
|
|
for identity in [dict(self.identity, bot_id='bot-b'), dict(self.identity, generation=2)]:
|
||
|
|
denied = self.call(dict(action='status', job_id=first['job_id'], identity=identity), check=False)
|
||
|
|
self.assertEqual(denied['error'], 'UNKNOWN_JOB')
|
||
|
|
done = self.wait(first['job_id'])
|
||
|
|
self.assertEqual(done['code'], 7)
|
||
|
|
self.assertEqual(base64.b64decode(done['stdout_base64']), b'\xffAB')
|
||
|
|
self.assertEqual(done['stderr'], 'error')
|
||
|
|
# Every call starts a new CLI process, proving output survives client reconnects.
|
||
|
|
rest = self.follow(first['job_id'], 'output', stdout_cursor=2, stderr_cursor=5)
|
||
|
|
self.assertEqual(rest['stdout'], 'B')
|
||
|
|
self.assertEqual(rest['stderr'], '')
|
||
|
|
self.assertEqual(rest['next_stdout_cursor'], 3)
|
||
|
|
self.assertEqual(self.follow(first['job_id'], 'cancel')['status'], 'failed')
|
||
|
|
|
||
|
|
def test_real_pty_input_and_resize(self):
|
||
|
|
_, started = self.start(['python3', '-u', '-c', 'import os;print(os.isatty(0));print(input().upper())'], pty=True)
|
||
|
|
job = started['job_id']
|
||
|
|
reply = self.follow(job, 'interact', input='hello 中文\n', rows=30, cols=100)
|
||
|
|
self.assertTrue(reply['pty'])
|
||
|
|
done = self.wait(job)
|
||
|
|
self.assertEqual(done['status'], 'succeeded')
|
||
|
|
self.assertIn('True', done['stdout'])
|
||
|
|
self.assertIn('HELLO 中文', done['stdout'])
|
||
|
|
|
||
|
|
def test_task_cannot_read_control_state_or_inherit_secrets(self):
|
||
|
|
os.environ['LAZYBOY_CONTROL_TOKEN'] = 'runner-secret-canary'
|
||
|
|
_, started = self.start(['python3', '-c', "import os; print(os.getuid()); print(os.getenv('LAZYBOY_CONTROL_TOKEN','unset')); open('/var/lib/lazyboy-runner/jobs.sqlite').read()"])
|
||
|
|
done = self.wait(started['job_id'])
|
||
|
|
self.assertEqual(done['status'], 'failed')
|
||
|
|
self.assertIn('1000\nunset', done['stdout'])
|
||
|
|
self.assertIn('PermissionError', done['stderr'])
|
||
|
|
self.assertNotIn('runner-secret-canary', json.dumps(done))
|
||
|
|
denied = subprocess.run(['runuser', '-u', 'lazyboy', '--', 'cat', RUNNER], capture_output=True)
|
||
|
|
self.assertNotEqual(denied.returncode, 0)
|
||
|
|
|
||
|
|
def test_pause_resume_cancel_are_scoped_to_one_job(self):
|
||
|
|
_, first = self.start(['sleep', '30'])
|
||
|
|
_, peer = self.start(['sh', '-c', 'sleep .3; printf peer'])
|
||
|
|
job = first['job_id']
|
||
|
|
self.assertEqual(self.follow(job, 'pause')['status'], 'paused')
|
||
|
|
self.assertEqual(self.wait(peer['job_id'])['stdout'], 'peer')
|
||
|
|
self.assertEqual(self.follow(job, 'resume')['status'], 'running')
|
||
|
|
self.follow(job, 'cancel')
|
||
|
|
self.assertEqual(self.wait(job)['status'], 'cancelled')
|
||
|
|
|
||
|
|
def test_deadline_and_output_quota_stop_process_tree(self):
|
||
|
|
job = 'test-' + uuid.uuid4().hex
|
||
|
|
self.jobs.append(job)
|
||
|
|
self.call(dict(action='start', job_id=job, identity=self.identity,
|
||
|
|
argv=['sh', '-c', 'sleep 30 & wait'], runtime_timeout_ms=200))
|
||
|
|
self.assertEqual(self.wait(job)['status'], 'timed_out')
|
||
|
|
_, loud = self.start(['python3', '-c', "import sys; sys.stdout.buffer.write(b'x'*6000000)"])
|
||
|
|
done = self.wait(loud['job_id'])
|
||
|
|
self.assertEqual(done['status'], 'output_limit')
|
||
|
|
self.assertTrue(done['truncated'])
|
||
|
|
self.assertEqual(done['next_stdout_cursor'], 65536)
|
||
|
|
|
||
|
|
def test_local_outbox_requires_explicit_ack(self):
|
||
|
|
_, job = self.start(['true'])
|
||
|
|
self.wait(job['job_id'])
|
||
|
|
before = self.call({'action': 'events'})['events']
|
||
|
|
own = [event for event in before if event['job_id'] == job['job_id']]
|
||
|
|
self.assertTrue(any(event['event'] == 'job.accepted' for event in own))
|
||
|
|
self.assertTrue(any(event['event'] == 'job.succeeded' for event in own))
|
||
|
|
self.call({'action': 'ack_events', 'ids': [event['id'] for event in own]})
|
||
|
|
after = self.call({'action': 'events'})['events']
|
||
|
|
self.assertFalse(any(event['job_id'] == job['job_id'] for event in after))
|
||
|
|
|
||
|
|
|
||
|
|
def test_expired_output_keeps_replay_tombstone(self):
|
||
|
|
request, job = self.start(['printf', 'retained'])
|
||
|
|
self.wait(job['job_id'])
|
||
|
|
with sqlite3.connect('/var/lib/lazyboy-runner/jobs.sqlite') as db:
|
||
|
|
state = json.loads(db.execute('SELECT state FROM jobs WHERE id=?', (job['job_id'],)).fetchone()[0])
|
||
|
|
state['completed_at'] = time.time() - 90000
|
||
|
|
db.execute('UPDATE jobs SET state=? WHERE id=?', (json.dumps(state), job['job_id']))
|
||
|
|
# Undelivered terminal evidence prevents expiration, even after the TTL.
|
||
|
|
events = self.call({'action': 'events'})['events']
|
||
|
|
self.assertEqual(self.follow(job['job_id'])['stdout'], 'retained')
|
||
|
|
own = [event['id'] for event in events if event['job_id'] == job['job_id']]
|
||
|
|
self.call({'action': 'ack_events', 'ids': own})
|
||
|
|
self.call({'action': 'events'})
|
||
|
|
replay = self.call(request)
|
||
|
|
self.assertTrue(replay['output_expired'])
|
||
|
|
self.assertEqual(replay['status'], 'succeeded')
|
||
|
|
self.assertFalse(Path('/var/lib/lazyboy-runner', job['job_id']).exists())
|
||
|
|
denied = self.call(dict(action='output', job_id=job['job_id'], identity=self.identity), check=False)
|
||
|
|
self.assertEqual(denied['error'], 'OUTPUT_EXPIRED')
|
||
|
|
changed = self.call(dict(request, argv=['false']), check=False)
|
||
|
|
self.assertEqual(changed['error'], 'PAYLOAD_MISMATCH')
|
||
|
|
|
||
|
|
def test_worker_crash_recovery_checks_process_identity(self):
|
||
|
|
_, job = self.start(['sh', '-c', 'sleep 30 & wait'])
|
||
|
|
with sqlite3.connect('/var/lib/lazyboy-runner/jobs.sqlite') as db:
|
||
|
|
state = json.loads(db.execute('SELECT state FROM jobs WHERE id=?', (job['job_id'],)).fetchone()[0])
|
||
|
|
os.kill(state['worker_pid'], signal.SIGKILL)
|
||
|
|
time.sleep(.05)
|
||
|
|
self.assertEqual(self.follow(job['job_id'])['status'], 'unknown')
|
||
|
|
# A forged/reused starttime must never authorize killing the PID.
|
||
|
|
with sqlite3.connect('/var/lib/lazyboy-runner/jobs.sqlite') as db:
|
||
|
|
stale = json.loads(db.execute('SELECT state FROM jobs WHERE id=?', (job['job_id'],)).fetchone()[0])
|
||
|
|
stale['pid_start'] = 'impossible'
|
||
|
|
db.execute('UPDATE jobs SET state=? WHERE id=?', (json.dumps(stale), job['job_id']))
|
||
|
|
denied = self.call(dict(action='cancel', job_id=job['job_id'], identity=self.identity), check=False)
|
||
|
|
self.assertEqual(denied['error'], 'ORPHAN_IDENTITY_UNPROVEN')
|
||
|
|
os.kill(state['pid'], 0)
|
||
|
|
with sqlite3.connect('/var/lib/lazyboy-runner/jobs.sqlite') as db:
|
||
|
|
stale['pid_start'] = state['pid_start']
|
||
|
|
db.execute('UPDATE jobs SET state=? WHERE id=?', (json.dumps(stale), job['job_id']))
|
||
|
|
self.assertEqual(self.follow(job['job_id'], 'cancel')['status'], 'interrupted')
|
||
|
|
_, unattended = self.start(['sleep', '30'])
|
||
|
|
with sqlite3.connect('/var/lib/lazyboy-runner/jobs.sqlite') as db:
|
||
|
|
state = json.loads(db.execute('SELECT state FROM jobs WHERE id=?', (unattended['job_id'],)).fetchone()[0])
|
||
|
|
os.kill(state['worker_pid'], signal.SIGKILL)
|
||
|
|
time.sleep(.05)
|
||
|
|
self.call({'action': 'events'})
|
||
|
|
self.assertEqual(self.follow(unattended['job_id'])['status'], 'interrupted')
|
||
|
|
|
||
|
|
|
||
|
|
def test_retention_pressure_releases_slots_without_replaying(self):
|
||
|
|
prefix = 'retention-' + uuid.uuid4().hex
|
||
|
|
with sqlite3.connect('/var/lib/lazyboy-runner/jobs.sqlite') as db:
|
||
|
|
states = [json.loads(row[0]) for row in db.execute('SELECT state FROM jobs')]
|
||
|
|
retained = sum(not state.get('output_expired') for state in states)
|
||
|
|
for index in range(64 - retained):
|
||
|
|
job = prefix + str(index)
|
||
|
|
state = dict(job_id=job, status='succeeded', code=0, identity=self.identity,
|
||
|
|
completed_at=time.time() - 60, created_at=time.time() - 61,
|
||
|
|
stdout_size=0, stderr_size=0, pty=False)
|
||
|
|
db.execute('INSERT INTO jobs VALUES (?,?,?,?)', (job, 'digest', '{}', json.dumps(state)))
|
||
|
|
try:
|
||
|
|
_, started = self.start(['true'])
|
||
|
|
self.assertEqual(self.wait(started['job_id'])['status'], 'succeeded')
|
||
|
|
with sqlite3.connect('/var/lib/lazyboy-runner/jobs.sqlite') as db:
|
||
|
|
states = [json.loads(row[0]) for row in db.execute('SELECT state FROM jobs')]
|
||
|
|
self.assertLessEqual(sum(not state.get('output_expired') for state in states), 64)
|
||
|
|
self.assertTrue(any(state.get('output_expired') and state['job_id'].startswith(prefix) for state in states))
|
||
|
|
finally:
|
||
|
|
with sqlite3.connect('/var/lib/lazyboy-runner/jobs.sqlite') as db:
|
||
|
|
db.execute('DELETE FROM jobs WHERE id LIKE ?', (prefix + '%',))
|
||
|
|
|
||
|
|
|
||
|
|
def test_agent_barrier_blocks_admission_and_preserves_peer(self):
|
||
|
|
_, own = self.start(['sleep', '30'])
|
||
|
|
_, manual = self.start(['sleep', '30'])
|
||
|
|
self.follow(manual['job_id'], 'pause')
|
||
|
|
peer_identity = dict(self.identity, bot_id='bot-b')
|
||
|
|
peer = 'peer-' + uuid.uuid4().hex
|
||
|
|
self.call(dict(action='start', job_id=peer, identity=peer_identity,
|
||
|
|
argv=['sh', '-c', 'sleep .2; printf peer-finished'], runtime_timeout_ms=5000))
|
||
|
|
try:
|
||
|
|
paused = self.call(dict(action='pause_agent', identity=self.identity))
|
||
|
|
self.assertEqual(paused['status'], 'paused')
|
||
|
|
self.assertEqual(self.follow(own['job_id'])['status'], 'paused')
|
||
|
|
rejected = self.call(dict(action='start', job_id='blocked-' + uuid.uuid4().hex,
|
||
|
|
identity=self.identity, argv=['true']), check=False)
|
||
|
|
self.assertEqual(rejected['error'], 'SCOPE_PAUSED')
|
||
|
|
bypass = self.call(dict(action='resume', job_id=own['job_id'], identity=self.identity), check=False)
|
||
|
|
self.assertEqual(bypass['error'], 'SCOPE_PAUSED')
|
||
|
|
time.sleep(.25)
|
||
|
|
other = self.call(dict(action='status', job_id=peer, identity=peer_identity))
|
||
|
|
self.assertEqual(other['status'], 'succeeded')
|
||
|
|
self.assertEqual(other['stdout'], 'peer-finished')
|
||
|
|
finally:
|
||
|
|
self.call(dict(action='resume_agent', identity=self.identity))
|
||
|
|
self.call(dict(action='cancel', job_id=peer, identity=peer_identity))
|
||
|
|
self.assertEqual(self.follow(own['job_id'])['status'], 'running')
|
||
|
|
self.assertEqual(self.follow(manual['job_id'])['status'], 'paused')
|
||
|
|
|
||
|
|
|
||
|
|
if __name__ == '__main__':
|
||
|
|
unittest.main()
|