193 lines
8.0 KiB
Python
193 lines
8.0 KiB
Python
#!/usr/bin/env python3
|
|
"""Full release-manifest CLI; synthetic inert artifacts, never installed."""
|
|
import io
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import resource
|
|
import shutil
|
|
import subprocess
|
|
import sys
|
|
import tarfile
|
|
import unittest
|
|
|
|
SOURCE = Path(os.environ['PACKAGER_SOURCE']).resolve()
|
|
WORK = Path(os.environ['PACKAGER_WORK']).resolve()
|
|
WORK.mkdir(parents=True, exist_ok=True)
|
|
RECEIPTS = []
|
|
REQUIRED = (
|
|
'scripts/ota-crash-guard.sh',
|
|
'image-recipe/configs/prepare-terminal-workspace.sh',
|
|
'scripts/d09-lifecycle-client.py',
|
|
'scripts/d09-lifecycle-owner.py',
|
|
'scripts/provision-d09-lifecycle.py',
|
|
'scripts/d09_wire.py',
|
|
'scripts/d09_store.py',
|
|
'scripts/d09_worker.py',
|
|
'scripts/d09-wire-schema.json',
|
|
'image-recipe/configs/archipelago-d09-lifecycle.service',
|
|
'image-recipe/configs/50-d09-lifecycle.conf',
|
|
'image-recipe/configs/60-terminal-workspace.conf',
|
|
)
|
|
|
|
|
|
def limits():
|
|
resource.setrlimit(resource.RLIMIT_CPU, (15, 15))
|
|
resource.setrlimit(resource.RLIMIT_AS, (256 * 1024**2, 256 * 1024**2))
|
|
resource.setrlimit(resource.RLIMIT_FSIZE, (8 * 1024**2, 8 * 1024**2))
|
|
|
|
|
|
class Packager(unittest.TestCase):
|
|
def setUp(self):
|
|
self.root = WORK / self._testMethodName
|
|
self.root.mkdir()
|
|
self.project = self.root / 'project'
|
|
shutil.copytree(SOURCE, self.project)
|
|
cargo = self.project / 'core/archipelago/Cargo.toml'
|
|
cargo.parent.mkdir(parents=True, exist_ok=True)
|
|
cargo.write_text('[package]\nversion = "1.9.2"\n')
|
|
self.backend = self.root / 'inert-backend'
|
|
self.backend.write_bytes(b'fixture only; not executable\n')
|
|
self.archive = self.root / 'frontend.tar.gz'
|
|
self.output = self.root / 'manifest.json'
|
|
|
|
def make_archive(self, omit=(), changed=None, mode=0o644, member_type=tarfile.REGTYPE, duplicate=False, bad_parent=False, alias=False):
|
|
with tarfile.open(self.archive, 'w:gz') as tf:
|
|
for path in REQUIRED:
|
|
if path in omit:
|
|
continue
|
|
data = (self.project / path).read_bytes()
|
|
if path == REQUIRED[1] and changed is not None:
|
|
data = changed
|
|
m = tarfile.TarInfo('./archipelago-runtime/' + path)
|
|
m.mode = mode if path == REQUIRED[1] else 0o644
|
|
m.type = member_type if path == REQUIRED[1] else tarfile.REGTYPE
|
|
if m.type in (tarfile.SYMTYPE, tarfile.LNKTYPE):
|
|
m.linkname = 'foreign-helper'
|
|
m.size = len(data) if m.isreg() else 0
|
|
tf.addfile(m, io.BytesIO(data) if m.isreg() else None)
|
|
if duplicate and path == REQUIRED[1]:
|
|
m.name = m.name.removeprefix('./')
|
|
tf.addfile(m, io.BytesIO(data) if m.isreg() else None)
|
|
if bad_parent:
|
|
m = tarfile.TarInfo('./archipelago-runtime/image-recipe')
|
|
m.type = tarfile.SYMTYPE; m.linkname = 'foreign-directory'
|
|
tf.addfile(m)
|
|
if alias:
|
|
data = b'wrong helper behind alias'
|
|
m = tarfile.TarInfo('./archipelago-runtime/image-recipe/configs/./prepare-terminal-workspace.sh')
|
|
m.mode = 0o644; m.size = len(data)
|
|
tf.addfile(m, io.BytesIO(data))
|
|
data = b'<!doctype html><title>Synthetic fixture</title>'
|
|
m = tarfile.TarInfo('./index.html'); m.mode = 0o644; m.size = len(data)
|
|
tf.addfile(m, io.BytesIO(data))
|
|
|
|
def invoke(self):
|
|
cmd = ['/bin/bash', str(self.project / 'scripts/create-release-manifest.sh'),
|
|
'--version', '1.9.2', '--date', '2026-10-25', '--backend', str(self.backend),
|
|
'--frontend', str(self.archive), '--output', str(self.output),
|
|
'--base-url', 'https://invalid.example/fixture']
|
|
env = {'PATH': '/usr/bin:/bin', 'HOME': str(self.root), 'TMPDIR': str(self.root),
|
|
'LANG': 'C', 'PYTHONDONTWRITEBYTECODE': '1'}
|
|
p = subprocess.run(cmd, cwd=self.root, env=env, capture_output=True, text=True,
|
|
timeout=10, preexec_fn=limits)
|
|
RECEIPTS.append({'case': self._testMethodName, 'command': cmd, 'exit': p.returncode,
|
|
'stdout': p.stdout, 'stderr': p.stderr, 'manifest_exists': self.output.exists()})
|
|
return p
|
|
|
|
def test_exact_helpers_emit_manifest(self):
|
|
self.make_archive()
|
|
p = self.invoke()
|
|
self.assertEqual(p.returncode, 0, p.stderr)
|
|
manifest = json.loads(self.output.read_text())
|
|
self.assertEqual(len(manifest['components']), 2)
|
|
import hashlib
|
|
self.assertEqual(manifest['components'][1]['sha256'], hashlib.sha256(self.archive.read_bytes()).hexdigest())
|
|
|
|
def test_gnu_tar_runtime_layout_emits_manifest(self):
|
|
staging = self.root / 'staging'
|
|
for path in REQUIRED:
|
|
dest = staging / 'archipelago-runtime' / path
|
|
dest.parent.mkdir(parents=True, exist_ok=True)
|
|
shutil.copyfile(self.project / path, dest); dest.chmod(0o644)
|
|
(staging / 'index.html').write_text('synthetic frontend')
|
|
staging.chmod(0o755)
|
|
for p in staging.rglob('*'):
|
|
if p.is_dir(): p.chmod(0o755)
|
|
cmd = ['/usr/bin/tar', '--owner=0', '--group=0', '--mode=u=rwX,go=rX',
|
|
'-czf', str(self.archive), '-C', str(staging), '.']
|
|
subprocess.run(cmd, check=True, timeout=10, preexec_fn=limits)
|
|
p = self.invoke()
|
|
self.assertEqual(p.returncode, 0, p.stderr)
|
|
self.assertTrue(self.output.exists())
|
|
|
|
def test_changed_helper_refuses_manifest(self):
|
|
original = (self.project / REQUIRED[1]).read_bytes()
|
|
self.make_archive(changed=bytes([original[0] ^ 1]) + original[1:])
|
|
p = self.invoke()
|
|
self.assertNotEqual(p.returncode, 0, 'real packager accepted wrong helper bytes')
|
|
self.assertIn('source mismatch', p.stderr)
|
|
self.assertFalse(self.output.exists())
|
|
|
|
def test_truncated_helper_refuses(self):
|
|
self.make_archive(changed=b'# truncated helper\n')
|
|
self.assert_refused()
|
|
|
|
def assert_refused(self):
|
|
p = self.invoke()
|
|
self.assertNotEqual(p.returncode, 0, 'real packager emitted invalid lifecycle payload')
|
|
self.assertIn('Error:', p.stderr)
|
|
self.assertFalse(self.output.exists())
|
|
|
|
def test_nonreadable_helper_refuses(self):
|
|
self.make_archive(mode=0o600)
|
|
self.assert_refused()
|
|
|
|
def test_duplicate_helper_refuses(self):
|
|
self.make_archive(duplicate=True)
|
|
self.assert_refused()
|
|
|
|
def test_link_helper_refuses(self):
|
|
self.make_archive(member_type=tarfile.SYMTYPE)
|
|
self.assert_refused()
|
|
|
|
def test_source_symlink_refuses(self):
|
|
self.make_archive()
|
|
src = self.project / REQUIRED[1]
|
|
other = src.with_name('foreign-helper'); src.rename(other); src.symlink_to(other.name)
|
|
self.assert_refused()
|
|
|
|
def test_missing_crash_guard_refuses(self):
|
|
self.make_archive(omit=(REQUIRED[0],))
|
|
self.assert_refused()
|
|
|
|
def test_ancestor_link_refuses(self):
|
|
self.make_archive(bad_parent=True)
|
|
self.assert_refused()
|
|
|
|
def test_helper_path_alias_refuses(self):
|
|
self.make_archive(alias=True)
|
|
self.assert_refused()
|
|
|
|
def test_source_fifo_refuses_without_blocking(self):
|
|
self.make_archive()
|
|
src = self.project / REQUIRED[1]; src.unlink(); os.mkfifo(src)
|
|
self.assert_refused()
|
|
|
|
def test_missing_terminal_helper_refuses_manifest(self):
|
|
self.make_archive(omit=(REQUIRED[1],))
|
|
p = self.invoke()
|
|
self.assertNotEqual(p.returncode, 0, 'real packager accepted missing OTA helper')
|
|
self.assertFalse(self.output.exists())
|
|
|
|
def test_missing_shared_schema_refuses_manifest(self):
|
|
self.make_archive(omit=('scripts/d09-wire-schema.json',))
|
|
self.assert_refused()
|
|
|
|
|
|
if __name__ == '__main__':
|
|
try:
|
|
unittest.main(verbosity=2)
|
|
finally:
|
|
(WORK / 'commands.json').write_text(json.dumps(RECEIPTS, indent=2) + '\n')
|