343 lines
18 KiB
Python
343 lines
18 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_staging_filters_private_apps_without_changing_source_or_public_apps(self):
|
|
for domain in ('apps', 'docker'):
|
|
for name in ('kithmoot', 'V4V', 'node-demo-v4v', 'public-app'):
|
|
original = self.source / domain / name / 'retained.txt'
|
|
original.parent.mkdir(parents=True)
|
|
original.write_text('source must remain')
|
|
staged = self.dist / 'archipelago-runtime' / domain / name / 'retained.txt'
|
|
staged.parent.mkdir(parents=True)
|
|
staged.write_bytes(original.read_bytes())
|
|
catalog = self.dist / 'catalog.json'
|
|
catalog.write_text(json.dumps({'apps': [{'id': 'kithmoot'}, {'id': 'public-app'}],
|
|
'storefront': {'promotions': [{'appId': 'v4v'}]}}))
|
|
payload.stage(self.source, self.dist, VERSION, COMMIT)
|
|
for domain in ('apps', 'docker'):
|
|
for name in ('kithmoot', 'V4V', 'node-demo-v4v'):
|
|
self.assertFalse((self.dist / 'archipelago-runtime' / domain / name).exists())
|
|
self.assertEqual((self.source / domain / name / 'retained.txt').read_text(), 'source must remain')
|
|
self.assertTrue((self.dist / 'archipelago-runtime' / domain / 'public-app/retained.txt').exists())
|
|
self.assertEqual(json.loads(catalog.read_bytes()), {'apps': [{'id': 'public-app'}], 'storefront': {'promotions': []}})
|
|
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 NativePayloadTests(PayloadTests):
|
|
def setUp(self):
|
|
super().setUp()
|
|
shell = self.source/'aiui/packages/app/src/NativeShell.vue'
|
|
shell.parent.mkdir(parents=True); shell.write_text('native fixture')
|
|
(self.source/'scripts/native-helper.js').write_text('helper fixture')
|
|
payload.stage(self.source, self.dist, VERSION, COMMIT)
|
|
shutil.rmtree(self.dist/'aiui')
|
|
(self.dist/'assets').mkdir()
|
|
chunk = self.dist/'assets/NativeShell-fixture.js'; chunk.write_bytes(b'native chunk')
|
|
self.info = dict(schema=1,mode='native',source_commit=COMMIT,source_dirty=False,
|
|
dynamic_entry=True,native_chunk='assets/NativeShell-fixture.js',sha256=payload.digest(chunk.read_bytes()))
|
|
self.write_info()
|
|
|
|
def write_info(self):
|
|
(self.dist/'NATIVE-AIUI.json').write_text(json.dumps(self.info))
|
|
|
|
def test_aiui_previous_commit_rejected(self):
|
|
self.info['source_commit']='b'*40; self.write_info()
|
|
with self.assertRaisesRegex(ValueError,'native AIUI'): self.check()
|
|
|
|
def test_native_chunk_tamper_and_dirty_source_refused(self):
|
|
payload.check_ui(self.source,self.dist,COMMIT)
|
|
for key,value in [('source_dirty',True),('dynamic_entry',False),('native_chunk','../escape')]:
|
|
original=self.info[key]; self.info[key]=value; self.write_info()
|
|
with self.assertRaises(ValueError): self.check()
|
|
self.info[key]=original
|
|
self.write_info()
|
|
(self.dist/'assets/NativeShell-fixture.js').write_bytes(b'tampered')
|
|
with self.assertRaisesRegex(ValueError,'native AIUI'): self.check()
|
|
|
|
def test_native_rejects_standalone_bootstrap(self):
|
|
(self.dist/'aiui').mkdir(); (self.dist/'aiui/index.html').write_text('stale standalone')
|
|
with self.assertRaisesRegex(ValueError,'standalone AIUI'): self.check()
|
|
|
|
|
|
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)
|
|
|
|
public_output = self.root / 'public-manifest.json'
|
|
public_command = command[:-2]
|
|
public_command[public_command.index('--output') + 1] = str(public_output)
|
|
public_command.append('--application-only')
|
|
public_result = subprocess.run(public_command, env=env, capture_output=True, text=True)
|
|
self.assertEqual(public_result.returncode, 0, public_result.stdout + public_result.stderr)
|
|
public = json.loads(public_output.read_text())
|
|
self.assertEqual(public['application_update'], {'version': 2, 'mode': 'signed-application-runtime'})
|
|
self.assertNotIn('os_update', public)
|
|
self.assertEqual(len(public['components']), 2)
|
|
payload.validate_release_components(public)
|
|
self.assertEqual(public['components'], manifest['components'][:2])
|
|
mixed = dict(public, os_update=manifest['os_update'])
|
|
with self.assertRaises(ValueError):
|
|
payload.validate_release_components(mixed)
|
|
boolean_policy = json.loads(json.dumps(public))
|
|
boolean_policy['application_update']['version'] = True
|
|
with self.assertRaises(ValueError):
|
|
payload.validate_release_components(boolean_policy)
|
|
missing_policy = dict(public)
|
|
missing_policy.pop('application_update')
|
|
with self.assertRaises(ValueError):
|
|
payload.validate_release_components(missing_policy)
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|