160 lines
7.3 KiB
Python
160 lines
7.3 KiB
Python
#!/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_keeps_provenance_pending_and_does_not_embed_test_binary(self):
|
|
contract_path = ROOT / "apps/qortal/empty-cache-overlay-contract.json"
|
|
contract = json.loads(contract_path.read_text(encoding="utf-8"))
|
|
self.assertEqual(contract["upstream_provenance_status"], "pending")
|
|
self.assertIsNone(contract["provenance_receipt_sha256"])
|
|
dockerfile = (ROOT / "apps/qortal/Dockerfile").read_text(encoding="utf-8")
|
|
self.assertIn("prepare-runtime-jar.py", dockerfile)
|
|
self.assertIn("empty-cache-overlay-contract.json", dockerfile)
|
|
self.assertNotIn("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()
|