diff --git a/.github/workflows/qualify-native-loopback-gameplay-overlay.yml b/.github/workflows/qualify-native-loopback-gameplay-overlay.yml index a2084dabf..a9f7f1e00 100644 --- a/.github/workflows/qualify-native-loopback-gameplay-overlay.yml +++ b/.github/workflows/qualify-native-loopback-gameplay-overlay.yml @@ -9,6 +9,8 @@ on: - 'tests/test_native_vanilla263_mineflayer_causal.py' - 'scripts/qualify_native_vanilla263_anvil_blocks.py' - 'tests/test_native_vanilla263_anvil_blocks.py' + - 'scripts/qualify_native_vanilla263_anvil_guest_readback.py' + - 'tests/test_native_vanilla263_anvil_guest_readback.py' - '.github/workflows/qualify-native-loopback-gameplay-overlay.yml' workflow_dispatch: permissions: @@ -31,10 +33,13 @@ jobs: tests/test_native_vanilla263_loopback_player_overlay.py \ tests/test_native_vanilla263_mineflayer_causal.py \ scripts/qualify_native_vanilla263_anvil_blocks.py \ - tests/test_native_vanilla263_anvil_blocks.py + tests/test_native_vanilla263_anvil_blocks.py \ + scripts/qualify_native_vanilla263_anvil_guest_readback.py \ + tests/test_native_vanilla263_anvil_guest_readback.py python3 -m unittest -v tests.test_native_vanilla263_loopback_player_overlay python3 -m unittest -v tests.test_native_vanilla263_mineflayer_causal python3 -m unittest -v tests.test_native_vanilla263_anvil_blocks + python3 -m unittest -v tests.test_native_vanilla263_anvil_guest_readback node --check scripts/qualify_native_vanilla263_mineflayer_causal.js bash -n scripts/gha_kvm_truenas_rdte.sh - name: Preserve operational auth and physical TrueNAS hold diff --git a/scripts/qualify_native_vanilla263_anvil_guest_readback.py b/scripts/qualify_native_vanilla263_anvil_guest_readback.py new file mode 100644 index 000000000..3ac23496b --- /dev/null +++ b/scripts/qualify_native_vanilla263_anvil_guest_readback.py @@ -0,0 +1,66 @@ +#!/usr/bin/env python3 +"""Bounded G4 quiesced native world Anvil file readback seam (source-only). + +This is a fixed-path read-only adapter for the existing disposable native T6 +DDP session and its already admitted owned App/world. It has no network client, +credential issuance, provider mutation or disk export. Neither a synthetic +test nor a caller-provided bytes callback is an independent live guest receipt. +""" +from __future__ import annotations + +try: + from .qualify_native_vanilla263_anvil_blocks import ( + MAX_REGION, SCHEMA, inspect, require) + from .qualify_vanilla263_native_world_rollback import ( + APP, WORLD, app_state) +except ImportError: + from qualify_native_vanilla263_anvil_blocks import ( + MAX_REGION, SCHEMA, inspect, require) + from qualify_vanilla263_native_world_rollback import ( + APP, WORLD, app_state) + +PATH = WORLD + "/region/r.0.0.mca" +STAGES = ("candidate", "rollback") + + +def capture(*, call, wait_job, fetch_bytes, stage: str) -> dict: + """Only after existing *actual* guest runner verified owned App then stop. + + The runner MUST independently bind the current authenticated DDP session, + exact native service/world, player action, signed request, and App stop. + It must not publish the raw Anvil region, signed session or download URL. + """ + require(stage in STAGES, "BACKEND_STAGE_NOT_ADMITTED") + require(app_state(call) == "STOPPED", "BACKEND_QUIESCENCE_NOT_OBSERVED") + stat = call("filesystem.stat", [PATH]) + require(type(stat) is dict and stat.get("type") == "FILE" and + type(stat.get("size")) is int and + 8192 <= stat["size"] <= MAX_REGION, "BACKEND_REGION_FILE_NOT_EXACT") + row = call("core.download", [ + "filesystem.get", [PATH], "g4-owned-region-" + stage + ".mca", True]) + require(type(row) is list and len(row) == 2 and + type(row[0]) is int and row[0] > 0 and + type(row[1]) is str and + row[1].startswith("/_download/") and + len(row[1]) <= 300 and + ".." not in row[1] and "?" not in row[1] and "#" not in row[1], + "BACKEND_DOWNLOAD_ROUTE_INVALID") + wait_job(row[0], "G4 read-only owned Anvil file download") + blob = fetch_bytes(row[1]) + require(type(blob) is bytes and len(blob) == stat["size"], + "BACKEND_DOWNLOADED_FILE_DRIFT") + result = inspect(blob, stage) + # The host transfer API is separate from independent receipt admission: + # this module cannot prove the caller ran under authorized guest identity. + result.update({ + "classification": "SOURCE_ONLY_OWNED_ANVIL_DOWNLOAD_SEAM", + "downloaded_file_size": len(blob), + "requested_native_app": APP, + "requested_native_world_file_fixed": True, + "read_only_middleware_calls": True, + "live_guest_identity_independently_verified": False, + "backend_independent_world_oracle_pass": False, + "raw_region_published": False, + "next_gate": "EXACT_LIVE_GUEST_RUNNER_AND_INDEPENDENT_RECEIPT_BINDING", + }) + return result diff --git a/tests/test_native_vanilla263_anvil_guest_readback.py b/tests/test_native_vanilla263_anvil_guest_readback.py new file mode 100644 index 000000000..46f31604e --- /dev/null +++ b/tests/test_native_vanilla263_anvil_guest_readback.py @@ -0,0 +1,100 @@ +"""Stubbed middleware admission and negative surface tests; NOT a live guest.""" +from __future__ import annotations + +import unittest + +from scripts.qualify_native_vanilla263_anvil_guest_readback import capture, PATH +from scripts.qualify_native_vanilla263_anvil_blocks import Refused +from scripts.qualify_vanilla263_native_world_rollback import APP +from tests.test_native_vanilla263_anvil_blocks import build + + +class NativeReadbackTests(unittest.TestCase): + def setUp(self): + self.blob = build("candidate") + self.state = "STOPPED" + self.stat = {"type": "FILE", "size": len(self.blob)} + self.token = "/_download/synthetic-test-opaque" + self.record = [] + self.downloaded = [] + + def call(self, name, args): + self.record.append((name, args)) + if name == "app.query": + self.assertEqual(args, [[["id", "=", APP]]]) + return [{"name": APP, "state": self.state}] + if name == "filesystem.stat": + self.assertEqual(args, [PATH]) + return self.stat + if name == "core.download": + self.assertEqual(args, ["filesystem.get", [PATH], + "g4-owned-region-candidate.mca", True]) + return [1001, self.token] + self.fail("unexpected middleware RPC: " + name) + + def wait(self, job, label): + self.downloaded.append((job, label)) + + def fetch(self, token): + self.assertEqual(token, self.token) + return self.blob + + def capture_stage(self, stage="candidate"): + return capture(call=self.call, wait_job=self.wait, + fetch_bytes=self.fetch, stage=stage) + + def reject(self, reason): + with self.assertRaises(Refused) as cm: + self.capture_stage() + self.assertEqual(str(cm.exception), reason) + + def test_fixed_native_quiesced_read_only_path_and_nonclaim(self): + result = self.capture_stage() + self.assertEqual(result["classification"], + "SOURCE_ONLY_OWNED_ANVIL_DOWNLOAD_SEAM") + self.assertEqual(result["downloaded_file_size"], len(self.blob)) + self.assertTrue(result["requested_native_world_file_fixed"]) + self.assertFalse(result["live_guest_identity_independently_verified"]) + self.assertFalse(result["backend_independent_world_oracle_pass"]) + self.assertFalse(result["raw_region_published"]) + self.assertEqual(result["physical_true_nas_apply"], "HOLD") + self.assertEqual([name for name, _ in self.record], + ["app.query", "filesystem.stat", "core.download"]) + self.assertEqual(self.downloaded, [(1001, "G4 read-only owned Anvil file download")]) + + def test_wrong_stage_has_zero_provider_calls(self): + with self.assertRaisesRegex(Refused, "BACKEND_STAGE_NOT_ADMITTED"): + self.capture_stage("other") + self.assertEqual(self.record, []) + + def test_refuse_running_app_before_file_read(self): + self.state = "RUNNING" + self.reject("BACKEND_QUIESCENCE_NOT_OBSERVED") + self.assertEqual([name for name, _ in self.record], ["app.query"]) + + def test_refuse_missing_or_unsafe_file_stat(self): + for attr in ({"type": "DIRECTORY", "size": len(self.blob)}, + {"type": "FILE", "size": 0}, {"type": "FILE", "size": True}, + None): + with self.subTest(stat=attr): + self.stat = attr + self.reject("BACKEND_REGION_FILE_NOT_EXACT") + + def test_refuse_injected_redirect_or_query(self): + for path in ("https://foreign.invalid/a", "/_download/../secrets", + "/_download/a?x=secret", "/_download/a#fragment", "/outside"): + with self.subTest(path=path): + self.token = path + self.reject("BACKEND_DOWNLOAD_ROUTE_INVALID") + + def test_refuse_wrong_file_bytes(self): + self.blob = self.blob[:-1] + self.reject("BACKEND_DOWNLOADED_FILE_DRIFT") + + def test_refuse_wrong_effect_inside_download(self): + self.blob = build("rollback") + self.reject("SERVER_BLOCK_CONSEQUENCE_ABSENT") + + +if __name__ == "__main__": + unittest.main()