262 lines
13 KiB
Python
262 lines
13 KiB
Python
"""Disposable artifact tests: exercise byte validation, never privileged runtime."""
|
|
import ast
|
|
import importlib.util
|
|
import io
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import shutil
|
|
import struct
|
|
import subprocess
|
|
import sys
|
|
import tarfile
|
|
import tempfile
|
|
import unittest
|
|
|
|
SCRIPTS = Path(__file__).resolve().parents[1]
|
|
sys.path.insert(0, str(SCRIPTS))
|
|
import release_payload as payload
|
|
|
|
VERSION = '2.0.1-alpha'
|
|
COMMIT = 'a' * 40
|
|
|
|
|
|
class PayloadTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.tmp = tempfile.TemporaryDirectory()
|
|
self.addCleanup(self.tmp.cleanup)
|
|
self.root = Path(self.tmp.name)
|
|
self.source = self.root / 'source'
|
|
self.dist = self.root / 'dist'
|
|
for name, raw in {
|
|
'scripts/d09_handoff.py': b'handoff',
|
|
'scripts/c01_network_owner.py': b'network',
|
|
'scripts/host_future_capability.py': b'new typed helper',
|
|
'image-recipe/configs/archipelago-c01-network.service': b'unit',
|
|
'neode-ui/public/packages/archipelago-companion.apk': b'qualified APK bytes',
|
|
'neode-ui/public/packages/archipelago-companion.json': b'{"versionCode":58,"versionName":"0.5.38-debug"}',
|
|
}.items():
|
|
path = self.source / name
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
path.write_bytes(raw)
|
|
payload.stage(self.source, self.dist, VERSION, COMMIT)
|
|
for name, raw in {
|
|
'index.html': b'dashboard', 'aiui/index.html': b'AIUI',
|
|
'aiui/BUILD-INFO': f'commit={COMMIT}\nbase_path=/aiui/\n'.encode(),
|
|
**{'packages/' + p.name: p.read_bytes() for p in (self.source / 'neode-ui/public/packages').iterdir()},
|
|
}.items():
|
|
path = self.dist / name
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
path.write_bytes(raw)
|
|
|
|
def archive(self, omit=None, duplicate=None, prefix='', link=None):
|
|
path = self.root / 'frontend.tar.gz'
|
|
with tarfile.open(path, 'w:gz') as tar:
|
|
for source in sorted(self.dist.rglob('*')):
|
|
name = source.relative_to(self.dist).as_posix()
|
|
if name == omit:
|
|
continue
|
|
info = tar.gettarinfo(str(source), prefix + name)
|
|
info.mode = 0o755 if source.is_dir() else 0o644
|
|
if name == link:
|
|
info.type = tarfile.SYMTYPE
|
|
info.linkname = '/etc/passwd'
|
|
info.size = 0
|
|
tar.addfile(info, io.BytesIO(source.read_bytes()) if info.isreg() else None)
|
|
if name == duplicate:
|
|
tar.addfile(info, io.BytesIO(source.read_bytes()))
|
|
return path
|
|
|
|
def check(self, **kwargs):
|
|
payload.check_archive(self.source, self.archive(**kwargs), VERSION, COMMIT)
|
|
|
|
def test_complete_flat_archive(self):
|
|
self.check()
|
|
|
|
def test_new_helper_is_required_without_allowlist_update(self):
|
|
(self.source / 'scripts/host_concurrent.py').write_text('new lane helper')
|
|
with self.assertRaisesRegex(ValueError, 'missing payload'):
|
|
self.check()
|
|
|
|
def test_network_and_handoff_omissions_rejected(self):
|
|
for name in ('c01_network_owner.py', 'd09_handoff.py', 'host_future_capability.py'):
|
|
with self.subTest(name=name), self.assertRaisesRegex(ValueError, 'missing payload'):
|
|
self.check(omit='archipelago-runtime/scripts/' + name)
|
|
|
|
def test_stale_companion_metadata_and_apk_rejected(self):
|
|
for name in ('archipelago-companion.apk', 'archipelago-companion.json'):
|
|
path = self.dist / 'packages' / name
|
|
original = path.read_bytes()
|
|
path.write_bytes(b'old')
|
|
with self.subTest(name=name), self.assertRaisesRegex(ValueError, 'source mismatch'):
|
|
self.check()
|
|
path.write_bytes(original)
|
|
|
|
def test_duplicate_link_and_nested_layout_rejected(self):
|
|
for kwargs in ({'duplicate': 'index.html'}, {'link': 'index.html'}, {'prefix': 'neode-ui/'}):
|
|
with self.subTest(kwargs=kwargs), self.assertRaises(ValueError):
|
|
self.check(**kwargs)
|
|
|
|
def test_false_hash_declaration_rejected(self):
|
|
path = self.dist / 'archipelago-runtime/RELEASE-PAYLOAD.json'
|
|
value = json.loads(path.read_text())
|
|
value['version'] = '2.0.0-alpha'
|
|
path.write_text(json.dumps(value))
|
|
with self.assertRaisesRegex(ValueError, 'declaration mismatch'):
|
|
self.check()
|
|
|
|
def test_aiui_previous_commit_rejected(self):
|
|
(self.dist / 'aiui/BUILD-INFO').write_text('commit=' + 'b' * 40 + '\nbase_path=/aiui/\n')
|
|
with self.assertRaisesRegex(ValueError, 'AIUI'):
|
|
self.check()
|
|
|
|
def test_wrong_native_architecture_and_script_rejected(self):
|
|
path = self.root / 'backend'
|
|
header = bytearray(64)
|
|
header[:6] = b'\x7fELF\x02\x01'
|
|
struct.pack_into('<HH', header, 16, 3, 62)
|
|
path.write_bytes(header)
|
|
payload.check_native(path)
|
|
struct.pack_into('<H', header, 18, 183)
|
|
for raw in (header, b'#!/bin/sh\nexit 0\n'):
|
|
path.write_bytes(raw)
|
|
with self.assertRaises(ValueError):
|
|
payload.check_native(path)
|
|
|
|
def test_source_symlink_refused(self):
|
|
(self.source / 'scripts/host_link.py').symlink_to('/etc/passwd')
|
|
with self.assertRaisesRegex(ValueError, 'symlink'):
|
|
payload.stage(self.source, self.dist, VERSION, COMMIT)
|
|
|
|
|
|
class OwnerPackageTests(unittest.TestCase):
|
|
def test_provisioned_modules_include_local_import_dependencies(self):
|
|
# Inspect the production copy list without importing its root owner.
|
|
tree = ast.parse((SCRIPTS / 'provision-d09-lifecycle.py').read_text())
|
|
modules = next(ast.literal_eval(node.value) for node in tree.body
|
|
if isinstance(node, ast.Assign)
|
|
and any(isinstance(target, ast.Name) and target.id == 'MODULES'
|
|
for target in node.targets))
|
|
self.assertIn('c01_network_client.py', modules)
|
|
for name in modules:
|
|
if not name.endswith('.py'):
|
|
continue
|
|
for node in ast.walk(ast.parse((SCRIPTS / name).read_text())):
|
|
imports = ([alias.name for alias in node.names] if isinstance(node, ast.Import)
|
|
else [node.module] if isinstance(node, ast.ImportFrom) and node.module
|
|
else [])
|
|
for imported in imports:
|
|
dependency = imported.split('.')[0] + '.py'
|
|
if (SCRIPTS / dependency).is_file():
|
|
self.assertIn(dependency, modules, f'{name} imports uninstalled {dependency}')
|
|
|
|
def test_complete_repository_runtime_inventory(self):
|
|
files = payload.runtime_sources(SCRIPTS.parent)
|
|
for name in ('scripts/c01_endpoints.py', 'scripts/c01_network_client.py',
|
|
'image-recipe/configs/archipelago-c01-network.service'):
|
|
self.assertIn('archipelago-runtime/' + name, files)
|
|
|
|
def test_reproducible_complete_owner_archive_and_overwrite_refusal(self):
|
|
spec = importlib.util.spec_from_file_location('owner_package', SCRIPTS / 'package-d09-c01.py')
|
|
module = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(module)
|
|
with tempfile.TemporaryDirectory() as temp:
|
|
a, b = Path(temp) / 'a.tar', Path(temp) / 'b.tar'
|
|
module.package(a)
|
|
module.package(b)
|
|
self.assertEqual(a.read_bytes(), b.read_bytes())
|
|
with tarfile.open(a) as tar:
|
|
hashes = json.load(tar.extractfile('D09-C01-SOURCE-SHA256.json'))
|
|
for name in ('c01_network_owner.py', 'c01_network_client.py', 'c01_endpoints.py', 'd09_handoff.py'):
|
|
self.assertIn('usr/libexec/' + name, hashes)
|
|
for name, expected in hashes.items():
|
|
self.assertEqual(payload.digest(tar.extractfile(name).read()), expected)
|
|
with self.assertRaises(FileExistsError):
|
|
module.package(a)
|
|
|
|
|
|
class SourceTests(unittest.TestCase):
|
|
def test_tag_and_dirty_source_refusals(self):
|
|
with tempfile.TemporaryDirectory() as temp:
|
|
root = Path(temp)
|
|
def git(*args):
|
|
return subprocess.check_output(['git', '-C', str(root), *args], stderr=subprocess.DEVNULL, text=True).strip()
|
|
git('init')
|
|
(root / 'core/archipelago').mkdir(parents=True)
|
|
(root / 'neode-ui').mkdir()
|
|
(root / 'core/archipelago/Cargo.toml').write_text(f'[package]\nversion="{VERSION}"\n')
|
|
(root / 'neode-ui/package.json').write_text(json.dumps({'version': VERSION}))
|
|
git('add', '.')
|
|
git('-c', 'user.name=Fixture', '-c', 'user.email=fixture@example.invalid', 'commit', '-m', 'fixture')
|
|
git('tag', 'v' + VERSION)
|
|
with self.assertRaisesRegex(ValueError, 'annotated'):
|
|
payload.check_source(root, VERSION)
|
|
git('tag', '-d', 'v' + VERSION)
|
|
git('-c', 'user.name=Fixture', '-c', 'user.email=fixture@example.invalid', 'tag', '-a', 'v' + VERSION, '-m', 'fixture')
|
|
self.assertEqual(payload.check_source(root, VERSION), git('rev-parse', 'HEAD'))
|
|
(root / 'new-helper.py').write_text('unreviewed')
|
|
with self.assertRaisesRegex(ValueError, 'clean reviewed'):
|
|
payload.check_source(root, VERSION)
|
|
|
|
|
|
class ManifestIntegrationTests(unittest.TestCase):
|
|
setUp = PayloadTests.setUp
|
|
archive = PayloadTests.archive
|
|
|
|
def test_manifest_from_disposable_annotated_source_and_real_archive(self):
|
|
# Synthetic APK tools are explicit fixture programs, not production
|
|
# bypass flags. The actual staged APK is verified separately by the lane.
|
|
scripts = self.source / 'scripts'
|
|
for name in ('create-release-manifest.sh', 'release_payload.py'):
|
|
shutil.copyfile(SCRIPTS / name, scripts / name)
|
|
for name in ('ota-crash-guard.sh', 'd09-lifecycle-client.py', 'd09-lifecycle-owner.py',
|
|
'provision-d09-lifecycle.py', 'd09_wire.py', 'd09_store.py', 'd09_worker.py', 'd09-wire-schema.json'):
|
|
(scripts / name).write_text('fixture')
|
|
for name in ('os_update_plan.py', 'd09_wire.py', 'd09-wire-schema.json'):
|
|
shutil.copyfile(SCRIPTS / name, scripts / name)
|
|
configs = self.source / 'image-recipe/configs'
|
|
for name in ('prepare-terminal-workspace.sh', 'archipelago-d09-lifecycle.service',
|
|
'50-d09-lifecycle.conf', '60-terminal-workspace.conf'):
|
|
(configs / name).write_text('fixture')
|
|
(self.source / 'core/archipelago').mkdir(parents=True)
|
|
(self.source / 'core/archipelago/Cargo.toml').write_text(f'[package]\nversion="{VERSION}"\n')
|
|
(self.source / 'neode-ui/package.json').write_text(json.dumps({'version': VERSION}))
|
|
(self.source / 'CHANGELOG.md').write_text(f'## v{VERSION} (2026-11-01)\n\n- Synthetic artifact fixture\n')
|
|
def git(*args):
|
|
return subprocess.check_output(['git', '-C', str(self.source), *args], stderr=subprocess.DEVNULL, text=True).strip()
|
|
git('init'); git('add', '.')
|
|
identity = ('-c', 'user.name=Fixture', '-c', 'user.email=fixture@example.invalid')
|
|
git(*identity, 'commit', '-m', 'synthetic source')
|
|
git(*identity, 'tag', '-a', 'v' + VERSION, '-m', 'fixture')
|
|
commit = git('rev-parse', 'HEAD')
|
|
payload.stage(self.source, self.dist, VERSION, commit)
|
|
(self.dist / 'aiui/BUILD-INFO').write_text(f'commit={commit}\nbase_path=/aiui/\n')
|
|
archive = self.archive()
|
|
archive = archive.rename(self.root / f'archipelago-frontend-{VERSION}.tar.gz')
|
|
aapt, signer = self.root / 'aapt', self.root / 'apksigner'
|
|
aapt.write_text("#!/bin/sh\nprintf \"package: name='com.archipelago.app.debug' versionCode='58' versionName='0.5.38-debug'\\n\"\n")
|
|
signer.write_text("#!/bin/sh\nprintf '%s\\n' 'Verified using v1 scheme (fixture): true' 'Verified using v2 scheme (fixture): true' 'Verified using v3 scheme (fixture): true' 'Signer #1 certificate SHA-256 digest: d622e07e7f474246e8e2bcee828ca9bf6c6cdf9434c065c58fe89bb4eec2664d'\n")
|
|
aapt.chmod(0o755); signer.chmod(0o755)
|
|
output = self.root / 'manifest.json'
|
|
from test_os_update import candidate
|
|
plan = self.root / 'os-update-plan.json'
|
|
plan.write_text(json.dumps(candidate()))
|
|
command = ['bash', str(scripts / 'create-release-manifest.sh'), '--version', VERSION,
|
|
'--backend', '/bin/true', '--frontend', str(archive), '--output', str(output),
|
|
'--os-plan', str(plan)]
|
|
env = dict(os.environ, AAPT=str(aapt), APKSIGNER=str(signer), PYTHONDONTWRITEBYTECODE='1')
|
|
result = subprocess.run(command, env=env, capture_output=True, text=True)
|
|
self.assertEqual(result.returncode, 0, result.stdout + result.stderr)
|
|
manifest = json.loads(output.read_text())
|
|
self.assertEqual(manifest['source_commit'], commit)
|
|
self.assertEqual(len(manifest['components']), 3)
|
|
self.assertEqual(manifest['components'][1]['sha256'], payload.digest(archive.read_bytes()))
|
|
saved = output.read_bytes()
|
|
result = subprocess.run(command, env=env, capture_output=True, text=True)
|
|
self.assertNotEqual(result.returncode, 0)
|
|
self.assertEqual(output.read_bytes(), saved)
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|