#!/usr/bin/env python3 """Focused refusal/snapshot checks against pinned fixture inputs. Usage: python3 -B scripts/test-podsteadr-manifest-output.py PATH/fixture_manifests.py Generator and companion bytes are verified; generation reads a private checked configuration snapshot. Both input pins and repeated output hashes are printed. External source is read-only. All writes are synthetic children of TMPDIR. This is not a managed-app, key-binding, production-manifest or migration test. """ import contextlib import hashlib import importlib.util import io from pathlib import Path import sys import tempfile import unittest from unittest.mock import patch PIN = '0d61e839d4d2cb797c59130895a87bd50cb6e21327b930d1ed3aa27dc72ca13c' COMPANION_PIN = '654b30dc66322fc21294911b23b31f38acc2e4b2aa2f0b3831cb89e9936624c4' source = Path(sys.argv.pop(1)) @contextlib.contextmanager def load_fixture(source, companion=None): if source.is_symlink() or not source.is_file(): raise ValueError('Explicit regular source file required') raw = source.read_bytes() if hashlib.sha256(raw).hexdigest() != PIN: raise ValueError('Fixture generator differs from selected source pin') companion = companion if companion is not None else source.with_name('mediamtx-fixture.yml') if companion.is_symlink() or not companion.is_file(): raise ValueError('Explicit regular companion file required') config = companion.read_bytes() if hashlib.sha256(config).hexdigest() != COMPANION_PIN: raise ValueError('Fixture companion differs from selected configuration pin') # The generator must read the checked snapshot, not reopen the external path. # This is controlled fixture preparation, not a hostile-filesystem sandbox. with tempfile.TemporaryDirectory(prefix='podsteadr-checked-') as snapshot: snapshot_source = Path(snapshot) / source.name (Path(snapshot) / 'mediamtx-fixture.yml').write_bytes(config) spec = importlib.util.spec_from_loader('fixture_manifests_checked', loader=None) assert spec is not None checked = importlib.util.module_from_spec(spec) checked.__file__ = str(snapshot_source) exec(compile(raw, str(source), 'exec'), checked.__dict__) yield checked class OutputRefusals(unittest.TestCase): def setUp(self): self.tmp = tempfile.TemporaryDirectory() self.addCleanup(self.tmp.cleanup) self.root = Path(self.tmp.name) self.output = self.root / 'output' self.output.mkdir() def invoke(self, output=None, overrides=None): values = { 'app-image': 'a' * 64, 'mtx-image': 'b' * 64, 'blossom-image': 'c' * 64, 'server-pubkey': 'd' * 64, 'data-root': '/var/lib/archipelago/fixture-podsteadr-output-test', 'output': str(output if output is not None else self.output), } values.update(overrides or {}) argv = ['fixture'] for name, value in values.items(): argv.extend(['--' + name, value]) with patch.object(sys, 'argv', argv), contextlib.redirect_stdout(io.StringIO()): fixture.main() def test_changed_companion_is_refused_before_generation(self): companion = self.root / 'mediamtx-fixture.yml' companion.write_bytes(b'unverified configuration') with self.assertRaisesRegex(ValueError, 'companion'): with load_fixture(source, companion): self.fail('unverified companion accepted') self.assertEqual(list(self.output.iterdir()), []) def test_verified_companion_snapshot_survives_path_change(self): companion = self.root / 'mediamtx-fixture.yml' verified = source.with_name('mediamtx-fixture.yml').read_bytes() companion.write_bytes(verified) copied_source = self.root / 'fixture_manifests.py' copied_source.write_bytes(source.read_bytes()) with load_fixture(copied_source, companion) as checked: companion.write_bytes(b'unverified configuration') with patch.dict(globals(), fixture=checked): self.invoke() import json manifest = json.loads((self.output / 'podsteadr-mediamtx' / 'manifest.yml').read_text()) expected = verified.decode('utf-8').replace( 'http://127.0.0.1:8095/api/mediamtx/auth', 'http://podsteadr-app:8095/api/mediamtx/auth') expected = '# Fixture network aliases; not an installed first-start contract.\n' + '\n'.join( line for line in expected.splitlines() if not line.startswith('#')) + '\n' self.assertEqual(manifest['app']['files'][0]['content'], expected) def test_missing_or_symlink_companion_is_refused(self): missing = self.root / 'missing.yml' link = self.root / 'link.yml' link.symlink_to(source.with_name('mediamtx-fixture.yml')) for companion in (missing, link): with self.subTest(companion=companion.name): with self.assertRaisesRegex(ValueError, 'regular companion'): with load_fixture(source, companion): self.fail('nonregular companion accepted') def test_repeated_generation_matches_exact_output_bytes(self): import json second = self.root / 'second' second.mkdir() self.invoke() self.invoke(second) first_files = {str(path.relative_to(self.output)): path.read_bytes() for path in self.output.glob('*/manifest.yml')} second_files = {str(path.relative_to(second)): path.read_bytes() for path in second.glob('*/manifest.yml')} self.assertEqual(set(first_files), { 'podsteadr/manifest.yml', 'podsteadr-mediamtx/manifest.yml', 'podsteadr-blossom/manifest.yml'}) self.assertEqual(first_files, second_files) print(json.dumps({'generator_sha256': PIN, 'companion_sha256': COMPANION_PIN, 'fixture_output_sha256': {name: hashlib.sha256(raw).hexdigest() for name, raw in sorted(first_files.items())}}, sort_keys=True)) def test_output_symlink_preserves_empty_target(self): link = self.root / 'link' link.symlink_to(self.output, target_is_directory=True) with self.assertRaises(ValueError): self.invoke(link) self.assertEqual(list(self.output.iterdir()), []) self.assertTrue(link.is_symlink()) def test_non_directory_and_missing_output_are_not_replaced(self): file = self.root / 'file' file.write_bytes(b'preserve sentinel') with self.assertRaises(ValueError): self.invoke(file) self.assertEqual(file.read_bytes(), b'preserve sentinel') missing = self.root / 'missing' with self.assertRaises(ValueError): self.invoke(missing) self.assertFalse(missing.exists()) def test_even_empty_existing_child_prevents_partial_writes(self): child = self.output / 'podsteadr-mediamtx' child.mkdir() with self.assertRaises(ValueError): self.invoke() self.assertEqual(list(self.output.iterdir()), [child]) self.assertEqual(list(child.iterdir()), []) def test_invalid_inputs_refused_before_any_output(self): for field, value in ( ('app-image', 'sha256:' + 'a' * 63), ('mtx-image', 'b' * 64 + '\n'), ('blossom-image', 'C' * 64), ('server-pubkey', 'd' * 64 + '\n'), ('data-root', '/var/lib/archipelago/fixture-podsteadr-x/../operator'), ): with self.subTest(field=field): with self.assertRaises(ValueError): self.invoke(overrides={field: value}) self.assertEqual(list(self.output.iterdir()), []) def test_dangling_output_symlink_is_not_followed(self): target = self.root / 'missing-target' link = self.root / 'dangling' link.symlink_to(target, target_is_directory=True) with self.assertRaises(ValueError): self.invoke(link) self.assertTrue(link.is_symlink()) self.assertFalse(target.exists()) if __name__ == '__main__': with load_fixture(source) as fixture: unittest.main()