[G4 P0] Read fixed owned Anvil region via stopped disposable native App (source only)

.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

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

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()