mirror of
https://github.com/MacRimi/ProxMenux.git
synced 2026-11-01 10:22:08 +00:00
220 lines
12 KiB
Python
220 lines
12 KiB
Python
"""A completed OCI job needs real exit-zero proof, not stopped/CLI output."""
|
|
import contextlib
|
|
import copy
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import socket
|
|
import sys
|
|
import tempfile
|
|
import time
|
|
import unittest
|
|
from unittest.mock import patch
|
|
|
|
sys.path.insert(0, str(Path(__file__).resolve().parents[1]/'remote'))
|
|
import oci_compose_jobs as jobs
|
|
|
|
IID='55bb1839-f087-4b03-a9b8-75bec7537c9c'
|
|
DIGEST='sha256:'+'a'*64
|
|
IDENTITY={'vmid':4790,'installation_id':IID,'image_digest':DIGEST,'config_fingerprint':'f'*64}
|
|
PROOF={'raw_waitstatus':0,'exit_code':0,'fresh_start_observed':True,'protocol':'lxc-msg-native264'}
|
|
CONFIG=f'''description: OCI %3C!-- proxmenux-instance={IID} --%3E
|
|
rootfs: local-zfs:subvol-4790-disk-0,size=8G
|
|
entrypoint: sh -c 'exit 0'
|
|
env: TOKEN=literal-secret-value
|
|
unprivileged: 1
|
|
lxc.init.uid: 1000
|
|
lxc.init.gid: 1000
|
|
mp0: /mnt/test/shared,mp=/data,backup=0,size=8G
|
|
cores: 1
|
|
memory: 256
|
|
tags: temporary
|
|
'''
|
|
|
|
|
|
def pack(kind,name,value):return jobs.FRAME.pack(kind,str(name).encode(),value)
|
|
|
|
|
|
class FakeProcess:
|
|
def __init__(self,rc=255):self.returncode=rc
|
|
def poll(self):return self.returncode
|
|
def wait(self,timeout=None):return self.returncode
|
|
def terminate(self):self.returncode=-15
|
|
def kill(self):self.returncode=-9
|
|
|
|
|
|
class NativeExitTests(unittest.TestCase):
|
|
def attempt(self, events):
|
|
receiver,sender=socket.socketpair()
|
|
with receiver,sender:
|
|
receiver.setblocking(False)
|
|
@contextlib.contextmanager
|
|
def subscription():yield receiver,{'protocol':'test-native-frames'}
|
|
def start(*args,**kwargs):
|
|
sender.sendall(b''.join(pack(*event) for event in events))
|
|
return FakeProcess()
|
|
with patch.object(jobs,'_subscription',subscription), \
|
|
patch.object(jobs.subprocess,'Popen',side_effect=start), \
|
|
patch.object(jobs,'_status',return_value='stopped'):
|
|
return jobs._run_attempt(4790,time.monotonic()+.2)
|
|
|
|
def test_exit_zero_is_proven_even_when_pct_start_loses_the_fast_pid(self):
|
|
proof=self.attempt([(0,4790,1),(0,4790,2),(2,4790,0)])
|
|
self.assertEqual(proof['raw_waitstatus'],0)
|
|
self.assertEqual(proof['pct_start_returncode'],255)
|
|
|
|
def test_nonzero_and_sigkill_never_become_completed(self):
|
|
for raw in (7<<8,9):
|
|
with self.subTest(raw=raw),self.assertRaises(jobs.JobError):
|
|
self.attempt([(0,4790,1),(0,4790,2),(2,4790,raw)])
|
|
|
|
def test_stopped_or_missing_exit_is_not_success(self):
|
|
with self.assertRaisesRegex(jobs.JobError,'timed out'):
|
|
self.attempt([(0,4790,1),(0,4790,2),(0,4790,0)])
|
|
|
|
def test_exit_without_a_fresh_start_event_is_rejected(self):
|
|
with self.assertRaisesRegex(jobs.JobError,'fresh'):
|
|
self.attempt([(2,4790,0)])
|
|
|
|
def test_unrelated_guest_exit_zero_does_not_complete_this_job(self):
|
|
with self.assertRaisesRegex(jobs.JobError,'timed out'):
|
|
self.attempt([(0,4790,1),(2,123,0),(0,4790,0)])
|
|
|
|
def test_partial_frames_are_buffered_and_unknown_abi_is_rejected(self):
|
|
data=pack(2,4790,1792)
|
|
self.assertEqual(jobs._frames(data[:8]),([],data[:8]))
|
|
self.assertEqual(jobs._frames(data),([(2,'4790',1792)],b''))
|
|
with self.assertRaises(jobs.JobError):jobs._frames(pack(100,4790,0))
|
|
|
|
|
|
class IdentityAndTrustTests(unittest.TestCase):
|
|
def test_command_environment_user_and_mount_changes_invalidate_fingerprint(self):
|
|
first=jobs.config_fingerprint(CONFIG)
|
|
for before,after in [('exit 0','exit 7'),('literal-secret-value','changed'),
|
|
('lxc.init.uid: 1000','lxc.init.uid: 1001'),('/mnt/test/shared','/mnt/test/other')]:
|
|
with self.subTest(before=before):
|
|
self.assertNotEqual(jobs.config_fingerprint(CONFIG.replace(before,after)),first)
|
|
altered=CONFIG.replace('cores: 1','cores: 4').replace('memory: 256','memory: 4096').replace('tags: temporary','tags: renamed').replace('size=8G','size=16G')
|
|
self.assertEqual(jobs.config_fingerprint(altered),first)
|
|
self.assertNotIn('literal-secret-value',first)
|
|
|
|
def test_instance_uuid_digest_and_container_marker_are_bound(self):
|
|
record={'vmid':4790,'installation_id':IID,'observed':{'resolved_registry_digest':DIGEST}}
|
|
spec={'installation_id':IID,'image_digest':DIGEST}
|
|
def read(path):return json.dumps(record) if str(path).endswith('.json') else CONFIG
|
|
with patch.object(jobs,'_trusted_read',side_effect=read), patch.object(jobs,'_config_text',return_value=CONFIG):
|
|
result=jobs.current_identity(4790,spec)
|
|
self.assertEqual(result['installation_id'],IID)
|
|
with self.assertRaises(jobs.JobError):jobs.current_identity(4790,{**spec,'image_digest':'sha256:'+'b'*64})
|
|
with self.assertRaises(jobs.JobError):jobs.current_identity(4790,{**spec,'installation_id':str(__import__('uuid').uuid4())})
|
|
with self.assertRaises(jobs.JobError):jobs.current_identity(4790,{**spec,'expected_config_fingerprint':'b'*64})
|
|
with patch.object(jobs,'_trusted_read',return_value=json.dumps(record)), patch.object(jobs,'_config_text',return_value=CONFIG.replace(IID,'another-instance')):
|
|
with self.assertRaises(jobs.JobError):jobs.current_identity(4790,spec)
|
|
|
|
def test_staged_digest_requires_matching_child_and_stack_transactions(self):
|
|
txid='a'*32
|
|
stackid='b'*32
|
|
journal=jobs.INSTANCE_ROOT/'4790'/'transactions'/txid/'transaction.json'
|
|
parent=jobs.INSTANCE_ROOT/'4792'/'stack-transactions'/stackid/'transaction.json'
|
|
new_digest='sha256:'+'b'*64
|
|
record={'vmid':4790,'installation_id':IID,'status':'updating',
|
|
'observed':{'resolved_registry_digest':DIGEST},'transaction_id':txid,
|
|
'pending_transaction':str(journal),'pending_stack_transaction':str(parent)}
|
|
child={'schema_version':1,'id':txid,'vmid':4790,'phase':'candidate-installed',
|
|
'record':{'installation_id':IID},'registry_digest':new_digest,
|
|
'staged_config_sha256':__import__('hashlib').sha256(CONFIG.encode()).hexdigest(),
|
|
'coordinated':{'journal':str(parent),'id':'active-stack-operation'}}
|
|
stack={'schema_version':1,'id':'active-stack-operation','phase':'starting',
|
|
'plan':{'members':[{'vmid':4790,'installation_id':IID}]}}
|
|
def read(path):
|
|
return json.dumps(child if path==journal else stack if path==parent else record)
|
|
with patch.object(jobs,'_trusted_read',side_effect=read),patch.object(jobs,'_config_text',return_value=CONFIG):
|
|
spec=jobs.job_spec_from_record(record,CONFIG)
|
|
self.assertEqual(spec['image_digest'],new_digest)
|
|
self.assertEqual(jobs.current_identity(4790,spec)['image_digest'],new_digest)
|
|
child['staged_config_sha256']='0'*64
|
|
with self.assertRaises(jobs.JobError):jobs.current_identity(4790,spec)
|
|
child['staged_config_sha256']=__import__('hashlib').sha256(CONFIG.encode()).hexdigest()
|
|
stack['id']='another-operation'
|
|
with self.assertRaises(jobs.JobError):jobs.current_identity(4790,spec)
|
|
stack['id']='active-stack-operation';stack['phase']='replacing'
|
|
self.assertEqual(jobs.job_spec_from_record(record,CONFIG)['image_digest'],new_digest)
|
|
with self.assertRaises(jobs.JobError):jobs.current_identity(4790,spec)
|
|
|
|
def test_restore_generation_changes_identity_even_with_identical_config(self):
|
|
record={'vmid':4790,'installation_id':IID,'observed':{'resolved_registry_digest':DIGEST}}
|
|
spec={'installation_id':IID,'image_digest':DIGEST}
|
|
with patch.object(jobs,'_trusted_read',side_effect=lambda path:json.dumps(record)),patch.object(jobs,'_config_text',return_value=CONFIG):
|
|
before=jobs.current_identity(4790,spec)
|
|
record['recovered_at']='2026-10-28T12:00:00Z'
|
|
after=jobs.current_identity(4790,spec)
|
|
self.assertEqual(before['config_fingerprint'],after['config_fingerprint'])
|
|
self.assertNotEqual(before,after)
|
|
|
|
def test_metadata_final_symlinks_and_world_writes_are_rejected(self):
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
source=Path(directory)/'record';source.write_text('{}');source.chmod(0o600)
|
|
link=Path(directory)/'link';link.symlink_to(source)
|
|
self.assertEqual(jobs._trusted_read(source,owner=os.getuid()),'{}')
|
|
with self.assertRaises(OSError):jobs._trusted_read(link,owner=os.getuid())
|
|
source.chmod(0o666)
|
|
with self.assertRaises(jobs.JobError):jobs._trusted_read(source,owner=os.getuid())
|
|
|
|
def test_socket_name_is_the_source_verified_lxc_hash(self):
|
|
self.assertEqual(jobs._socket_name(),'lxc/ad055575fe28ddd5//var/lib/lxc')
|
|
|
|
|
|
class CompletionCacheTests(unittest.TestCase):
|
|
def setUp(self):
|
|
directory=tempfile.TemporaryDirectory();self.addCleanup(directory.cleanup)
|
|
self.root=Path(directory.name)
|
|
for name,replacement in [('STATE_ROOT',self.root),('_root',lambda:None),
|
|
('_job_lock',lambda *args:contextlib.nullcontext()),
|
|
('_status',lambda *args:'stopped')]:
|
|
item=patch.object(jobs,name,replacement);item.start();self.addCleanup(item.stop)
|
|
item=patch.object(jobs,'current_identity',return_value=IDENTITY);item.start();self.addCleanup(item.stop)
|
|
item=patch.object(jobs,'_read_cache',side_effect=lambda path:json.loads(path.read_text()) if path.exists() else None);item.start();self.addCleanup(item.stop)
|
|
self.spec={'installation_id':IID,'image_digest':DIGEST,'timeout_seconds':30}
|
|
|
|
def test_only_matching_proven_cache_skips_job_start(self):
|
|
with patch.object(jobs,'_run_attempt',return_value=PROOF) as attempt:
|
|
self.assertFalse(jobs.ensure_completed(4790,self.spec)['cached'])
|
|
self.assertTrue(jobs.ensure_completed(4790,self.spec)['cached'])
|
|
attempt.assert_called_once()
|
|
cache=json.loads((self.root/'4790.json').read_text())
|
|
self.assertEqual(cache['identity'],IDENTITY)
|
|
self.assertEqual(cache['proof']['raw_waitstatus'],0)
|
|
|
|
def test_changed_image_identity_or_config_never_reuses_old_completion(self):
|
|
cache={'schema':1,'status':'completed','identity':{**IDENTITY,'image_digest':'sha256:'+'b'*64},'proof':PROOF}
|
|
(self.root/'4790.json').write_text(json.dumps(cache))
|
|
with patch.object(jobs,'_run_attempt',return_value=PROOF) as attempt:
|
|
self.assertFalse(jobs.ensure_completed(4790,self.spec)['cached'])
|
|
attempt.assert_called_once()
|
|
|
|
def test_force_failure_invalidates_earlier_success(self):
|
|
with patch.object(jobs,'_run_attempt',return_value=PROOF):jobs.ensure_completed(4790,self.spec)
|
|
with patch.object(jobs,'_run_attempt',side_effect=jobs.JobError('signal 9')):
|
|
with self.assertRaises(jobs.JobError):jobs.ensure_completed(4790,self.spec,force=True)
|
|
self.assertFalse((self.root/'4790.json').exists())
|
|
|
|
def test_remove_forgets_only_matching_installation_uuid(self):
|
|
with patch.object(jobs,'_run_attempt',return_value=PROOF):jobs.ensure_completed(4790,self.spec)
|
|
other='45e85d70-4e28-416f-8ae8-325ae9153272'
|
|
self.assertFalse(jobs.forget_completed(4790,other))
|
|
self.assertTrue((self.root/'4790.json').exists())
|
|
self.assertTrue(jobs.forget_completed(4790,IID))
|
|
self.assertFalse((self.root/'4790.json').exists())
|
|
self.assertFalse(jobs.forget_completed(4790,IID))
|
|
|
|
def test_running_job_and_bad_timeout_are_refused(self):
|
|
with patch.object(jobs,'_status',return_value='running'),patch.object(jobs,'_run_attempt') as attempt:
|
|
with self.assertRaisesRegex(jobs.JobError,'already running'):jobs.ensure_completed(4790,self.spec)
|
|
attempt.assert_not_called()
|
|
for timeout in (True,0,29,3601,'300'):
|
|
with self.subTest(timeout=timeout),self.assertRaises(jobs.JobError):
|
|
jobs.ensure_completed(4790,{**self.spec,'timeout_seconds':timeout})
|
|
|
|
|
|
if __name__=='__main__':unittest.main()
|