#!/usr/bin/env python3 """Package-path checks for the guarded Qortal runtime overlay.""" import hashlib import json from pathlib import Path import subprocess import sys import tempfile import unittest import zipfile ROOT = Path(__file__).resolve().parents[3] HELPER = ROOT / "apps/qortal/prepare-runtime-jar.py" SETTINGS = ROOT / "apps/qortal/check-settings.py" TARGET = "org/qortal/controller/arbitrary/ArbitraryDataCacheManager.class" def sha(data: bytes) -> str: return hashlib.sha256(data).hexdigest() class PackageOverlayContract(unittest.TestCase): def setUp(self): self.tmp = tempfile.TemporaryDirectory() self.addCleanup(self.tmp.cleanup) self.root = Path(self.tmp.name) self.original = self.root / "qortal.jar" self.replacement = self.root / "ArbitraryDataCacheManager.class" self.contract = self.root / "overlay-contract.json" self.runtime = self.root / "runtime-qortal.jar" self.receipt = self.root / "runtime-qortal.json" self.data = self.root / "operator-data" self.data.mkdir() self.sentinel = self.data / "existing-chain-state" self.sentinel.write_bytes(b"nonempty operator state\n") self._write_jar() self.replacement.write_bytes(b"\xca\xfe\xba\xbeACCEPTED-EMPTY-CACHE-BEHAVIOR") self._write_contract("accepted") def _write_jar(self, target: bytes = b"\xca\xfe\xba\xbeOFFICIAL", other: bytes = b"preserve"): with zipfile.ZipFile(self.original, "w") as jar: jar.writestr(TARGET, target) jar.writestr("unchanged.txt", other) def _write_contract(self, status: str): doc = { "schema": "archipelago.qortal.empty-cache-overlay.v1", "qortal_version": "6.1.9", "upstream_tag": "v6.1.9", "upstream_commit": "108bf191d42d710ec617f535af30cfd82fc03c87", "original_jar_sha256": sha(self.original.read_bytes()), "target_entry": TARGET, "replacement_class_sha256": sha(self.replacement.read_bytes()), "source_patch_sha256": "7fccb0afc2113abe19af0e1fa2e70cb7a61fbd8c410ad372e4e9d28ef31ae0c8", "behavior_review_status": "accepted", "upstream_provenance_status": status, "provenance_receipt_sha256": "11" * 32 if status == "accepted" else None, } self.contract.write_text(json.dumps(doc), encoding="utf-8") def run_helper(self, mode: str): if mode == "prepare": args = [sys.executable, str(HELPER), "prepare", str(self.original), str(self.replacement), str(self.contract), str(self.runtime), str(self.receipt)] else: args = [sys.executable, str(HELPER), "rollback", str(self.original), str(self.runtime), str(self.receipt)] return subprocess.run(args, text=True, capture_output=True, check=False) def test_nonempty_upgrade_and_rollback_preserve_original_and_operator_data(self): original = self.original.read_bytes() sentinel = self.sentinel.read_bytes() prepared = self.run_helper("prepare") self.assertEqual(prepared.returncode, 0, prepared.stderr) self.assertEqual(self.original.read_bytes(), original) self.assertEqual(self.sentinel.read_bytes(), sentinel) with zipfile.ZipFile(self.runtime) as jar: self.assertEqual(jar.read(TARGET), self.replacement.read_bytes()) self.assertEqual(jar.read("unchanged.txt"), b"preserve") rolled_back = self.run_helper("rollback") self.assertEqual(rolled_back.returncode, 0, rolled_back.stderr) self.assertFalse(self.runtime.exists()) self.assertFalse(self.receipt.exists()) self.assertEqual(self.original.read_bytes(), original) self.assertEqual(self.sentinel.read_bytes(), sentinel) def test_pending_provenance_refuses_before_runtime_output(self): self._write_contract("pending") result = self.run_helper("prepare") self.assertEqual(result.returncode, 78) self.assertFalse(self.runtime.exists()) self.assertFalse(self.receipt.exists()) def test_missing_and_changed_original_jar_refuse(self): self.original.unlink() missing = self.run_helper("prepare") self.assertEqual(missing.returncode, 78) self.assertFalse(self.runtime.exists()) self._write_jar(other=b"changed official bytes") changed = self.run_helper("prepare") self.assertEqual(changed.returncode, 78) self.assertFalse(self.runtime.exists()) def test_changed_overlay_class_refuses(self): self.replacement.write_bytes(b"\xca\xfe\xba\xbeDRIFT") result = self.run_helper("prepare") self.assertEqual(result.returncode, 78) self.assertFalse(self.runtime.exists()) def test_rollback_refuses_changed_runtime_without_deleting_it(self): self.assertEqual(self.run_helper("prepare").returncode, 0) self.runtime.write_bytes(b"changed after preparation") result = self.run_helper("rollback") self.assertEqual(result.returncode, 78) self.assertTrue(self.runtime.exists()) self.assertTrue(self.receipt.exists()) self.assertEqual(self.sentinel.read_bytes(), b"nonempty operator state\n") def test_entrypoint_prepares_verified_runtime_before_settings_and_java(self): start = (ROOT / "apps/qortal/start.sh").read_text(encoding="utf-8") prepare = start.index("prepare-runtime-jar.py\" prepare") settings = start.index("qortal-check-settings") java = start.index("-jar \"$runtime_jar\"") self.assertLess(prepare, settings) self.assertLess(settings, java) self.assertNotIn("-jar /usr/local/qortal/qortal.jar", start) def test_package_binds_independently_accepted_inputs_and_corresponding_source(self): package = ROOT / "apps/qortal/package-input" contract = json.loads((ROOT / "apps/qortal/empty-cache-overlay-contract.json").read_text()) self.assertEqual(contract["upstream_provenance_status"], "accepted") self.assertEqual(contract["provenance_receipt_sha256"], "fbcf0edac758a8b669117743a9754d15a574f7f98da6ad17da2141ee09754b26") for name, expected in { "provenance-receipt.json": contract["provenance_receipt_sha256"], "inputs/ArbitraryDataCacheManager.class": "5fea2e7a6b61ec840fcc252e89032df26bdb84601e819932794e033d68d21c54", "source/qortal-v6.1.9-source.tar.gz": "e26935ea4069dcd058904d556e036ab95a2f9debbb02626535a963e6cadf8abc", "source/empty-cache.patch": contract["source_patch_sha256"], "source/QORTAL-LICENSE-GPL-3.0.txt": "3972dc9744f6499f0f9b2dbf76696f2ae7ad8af9b23dde66d6af86c9dfb36986", }.items(): with self.subTest(name=name): self.assertEqual(sha((package / name).read_bytes()), expected) self.assertTrue((package / "reproduce-class.sh").is_file()) self.assertIn("Archipelago changes ArbitraryDataCacheManager", (package / "NOTICE").read_text()) dockerfile = (ROOT / "apps/qortal/Dockerfile").read_text() self.assertIn("COPY package-input/ /usr/share/doc/qortal/", dockerfile) self.assertIn("sha256sum -c SHA256SUMS", dockerfile) self.assertIn("COPY package-input/inputs/ArbitraryDataCacheManager.class", dockerfile) def test_existing_credential_drift_refuses_through_package_validator(self): settings = self.root / "settings.json" snapshot = self.root / "snapshot.json" key = self.data / "apikey.txt" settings.write_text(json.dumps({ "localAuthBypassEnabled": False, "apiRestricted": True, "autoUpdateEnabled": False, }), encoding="utf-8") key.write_text("22" * 32, encoding="ascii") key.chmod(0o600) result = subprocess.run( [sys.executable, str(SETTINGS), str(settings), str(snapshot), str(key)], env={"QORTAL_API_KEY": "11" * 32}, text=True, capture_output=True, check=False) self.assertEqual(result.returncode, 78) self.assertEqual(key.read_text(encoding="ascii"), "22" * 32) self.assertFalse(snapshot.exists()) self.assertEqual(self.sentinel.read_bytes(), b"nonempty operator state\n") if __name__ == "__main__": unittest.main()