Add disabled exact-generation cloud input fetch for hosted reviews

.github/workflows/engineering-material-review.yml

@@ -25,18 +25,31 @@ jobs:
       - uses: actions/setup-python@v6
         with:
           python-version: '3.11'
+      - name: Fetch the approved cloud input bundle (disabled by default)
+        if: vars.HOSTED_INPUT_FETCH_ENABLED == 'true'
+        env:
+          HOSTED_INPUT_FETCH_ENABLED: ${{ vars.HOSTED_INPUT_FETCH_ENABLED }}
+          HOSTED_INPUT_URI: ${{ vars.HOSTED_INPUT_URI }}
+          HOSTED_INPUT_GENERATION: ${{ vars.HOSTED_INPUT_GENERATION }}
+          HOSTED_INPUT_ASSET: ${{ vars.HOSTED_INPUT_ASSET }}
+          HOSTED_INPUT_SHA256: ${{ vars.HOSTED_INPUT_SHA256 }}
+        run: python scripts/fetch_hosted_inputs.py
       - name: Stage the approved hosted input bundle
         env:
+          HOSTED_INPUT_FETCH_ENABLED: ${{ vars.HOSTED_INPUT_FETCH_ENABLED }}
           GH_TOKEN: ${{ github.token }}
           HOSTED_INPUT_RELEASE: ${{ vars.HOSTED_INPUT_RELEASE }}
           HOSTED_INPUT_ASSET: ${{ vars.HOSTED_INPUT_ASSET }}
           HOSTED_INPUT_SHA256: ${{ vars.HOSTED_INPUT_SHA256 }}
         run: |
           set -euo pipefail
-          test -n "$HOSTED_INPUT_RELEASE" && test -n "$HOSTED_INPUT_ASSET"
+          test -n "$HOSTED_INPUT_ASSET"
           [[ "$HOSTED_INPUT_SHA256" =~ ^[0-9a-f]{64}$ ]]
           [[ "$HOSTED_INPUT_ASSET" =~ ^[A-Za-z0-9_.-]+\.zip$ ]]
-          gh release download "$HOSTED_INPUT_RELEASE" --repo "$GITHUB_REPOSITORY" --pattern "$HOSTED_INPUT_ASSET" --dir "$RUNNER_TEMP/approved-input"
+          if [[ "$HOSTED_INPUT_FETCH_ENABLED" != true ]]; then
+            test -n "$HOSTED_INPUT_RELEASE"
+            gh release download "$HOSTED_INPUT_RELEASE" --repo "$GITHUB_REPOSITORY" --pattern "$HOSTED_INPUT_ASSET" --dir "$RUNNER_TEMP/approved-input"
+          fi
           python - <<'PYCODE'
           import hashlib, os, stat, zipfile
           from pathlib import Path, PurePosixPath

docs/hosted-ai-runners.md

@@ -16,8 +16,9 @@ config. Existing per-workflow material/source/wheel/root variables must point
 inside `hosted-inputs/` (output roots can use RUNNER_TEMP). Frozen source git
 metadata/commit pins and the existing wheel SHA256 checks still apply; do not
 replace a reviewed wheel with a similarly named package from an index.
-The bundle includes only licensed, approved research inputs and required
-reviewed runtime wheels. Publishing private inputs requires owner approval.
+Release assets must contain public code, synthetic material and reviewed
+runtime wheels only. Private material and real market/account inputs stay in
+approved cloud storage; do not publish them as repository release assets.
 
 The bundle must supply wheels/ and a fully pinned runtime-requirements.txt
 for the approved task-client distribution and dependencies. Offline pip uses
@@ -39,6 +40,36 @@ it does not call Dot or establish research/receipt completion.
 
 ## Acceptance and rollback
 
+### Optional cloud input fetch (draft; disabled by default)
+
+`HOSTED_INPUT_FETCH_ENABLED=true` selects a single object in approved GCS
+storage instead of a GitHub Release asset. It requires `HOSTED_INPUT_URI`
+(exact `gs://` object; no wildcards), `HOSTED_INPUT_GENERATION` (immutable
+generation), the existing `HOSTED_INPUT_SHA256` (archive SHA256), and
+`HOSTED_INPUT_ASSET` (local ZIP basename). There are no private default values.
+The fetcher runs only on a main-branch GitHub-hosted runner, checks the pinned
+object metadata before download, caps the compressed archive at 512 MiB,
+verifies its length and digest, and then uses the existing safe ZIP staging.
+It requests exact-object metadata/GET operations without listing the bucket.
+Fetch or validation failure stops the job; it does not fall back to a release
+asset, a different generation, or a server directory. CLI error details are
+withheld because they may contain private object names.
+
+The workflow does **not** configure GCP authentication. The `gcloud` CLI must
+already have an approved keyless identity on that hosted runner. Existing
+OIDC permission for the task service is not proof of GCS access. Before any
+enablement, review reuse of the existing GitHub OIDC/WIF pool/provider and a
+dedicated read-only service account restricted to the approved bucket prefix.
+Authentication, IAM, variables and secrets have not been changed here. The
+current review job has no GCP authentication step, so this remains a draft
+until that separate authorization and the actual bundle/resume acceptance
+are complete. Do not enable the flag to test missing permissions.
+
+Local tests use synthetic bytes and a fake cloud command. No real input is
+fetched to a developer machine, and CI does not authenticate, deploy, submit
+tasks, or call paid model APIs. Keep `HOSTED_RESEARCH_ACCEPTED` and
+`AI_SERVICE_RELEASE_READY` at their existing values.
+
 Run actionlint and the existing PR CI at the final commit. Before adoption,
 execute a synthetic/offline bundle and confirm wheel/input digest rejection,
 frozen-source identity, existing budgets and Docker availability where needed.

scripts/fetch_hosted_inputs.py

@@ -0,0 +1,99 @@
+"""Fetch one pinned cloud object on an approved GitHub-hosted runner only."""
+
+import hashlib
+import json
+import os
+from pathlib import Path
+import re
+import subprocess
+import tempfile
+
+
+MAX_ARCHIVE_BYTES = 512 * 1024**2
+
+
+def _gcloud(arguments):
+    try:
+        result = subprocess.run(
+            ["gcloud", "storage", *arguments, "--quiet"],
+            capture_output=True, text=True, timeout=180, check=False,
+        )
+    except (OSError, subprocess.TimeoutExpired):
+        raise ValueError("approved cloud input command unavailable or timed out") from None
+    if result.returncode:
+        # CLI errors can contain private object names or authentication details.
+        raise ValueError(f"approved cloud input command failed (exit {result.returncode})")
+    return result.stdout
+
+
+def fetch(environ=None):
+    env = os.environ if environ is None else environ
+    if env.get("HOSTED_INPUT_FETCH_ENABLED", "false") != "true":
+        return False
+    if (env.get("GITHUB_ACTIONS") != "true"
+            or env.get("RUNNER_ENVIRONMENT") != "github-hosted"
+            or env.get("GITHUB_REF") != "refs/heads/main"):
+        raise ValueError("cloud inputs require an approved main-branch GitHub-hosted run")
+
+    uri = env.get("HOSTED_INPUT_URI", "")
+    generation = env.get("HOSTED_INPUT_GENERATION", "")
+    expected_digest = env.get("HOSTED_INPUT_SHA256", "")
+    asset = env.get("HOSTED_INPUT_ASSET", "")
+    runner_temp = env.get("RUNNER_TEMP", "")
+    if not re.fullmatch(r"gs://[a-z0-9][a-z0-9._-]{1,220}[a-z0-9]/[A-Za-z0-9_./-]+", uri):
+        raise ValueError("HOSTED_INPUT_URI must name one exact GCS object without wildcards")
+    if not re.fullmatch(r"[1-9][0-9]*", generation):
+        raise ValueError("HOSTED_INPUT_GENERATION must pin a positive object generation")
+    if not re.fullmatch(r"[0-9a-f]{64}", expected_digest):
+        raise ValueError("HOSTED_INPUT_SHA256 must pin the approved archive digest")
+    if not re.fullmatch(r"[A-Za-z0-9_.-]+\.zip", asset):
+        raise ValueError("HOSTED_INPUT_ASSET must be a ZIP basename")
+    if not runner_temp or not Path(runner_temp).is_absolute():
+        raise ValueError("RUNNER_TEMP must be an absolute hosted-runner directory")
+
+    source = f"{uri}#{generation}"
+    try:
+        metadata = json.loads(_gcloud([
+            "objects", "describe", source, "--format=json(size,generation)",
+        ]))
+        size = int(metadata["size"])
+        actual_generation = str(metadata["generation"])
+    except (KeyError, TypeError, json.JSONDecodeError):
+        raise ValueError("approved cloud input metadata is invalid") from None
+    if actual_generation != generation or not 0 < size <= MAX_ARCHIVE_BYTES:
+        raise ValueError("approved cloud input generation or size rejected")
+
+    destination = Path(runner_temp) / "approved-input"
+    destination.mkdir(exist_ok=True)
+    archive = destination / asset
+    if destination.is_symlink() or archive.exists() or archive.is_symlink():
+        raise ValueError("approved cloud input destination must be fresh")
+    with tempfile.TemporaryDirectory(prefix="cloud-input-", dir=destination) as temporary:
+        partial = Path(temporary) / "bundle.zip"
+        _gcloud(["cp", source, str(partial)])
+        if partial.stat().st_size != size:
+            raise ValueError("approved cloud input size mismatch")
+        digest = hashlib.sha256()
+        with partial.open("rb") as stream:
+            for chunk in iter(lambda: stream.read(1024 * 1024), b""):
+                digest.update(chunk)
+        if digest.hexdigest() != expected_digest:
+            raise ValueError("approved cloud input digest mismatch")
+        partial.rename(archive)
+    return True
+
+
+def main():
+    try:
+        enabled = fetch()
+    except (OSError, ValueError) as error:
+        # Do not expose exception strings from file/CLI/JSON operations.
+        message = str(error) if isinstance(error, ValueError) else "approved cloud input file operation failed"
+        if not message.startswith(("approved cloud input", "cloud inputs require", "HOSTED_INPUT_", "RUNNER_TEMP")):
+            message = "approved cloud input validation failed"
+        raise SystemExit(message) from None
+    print("Approved cloud input verified" if enabled else "Cloud input fetch disabled")
+
+
+if __name__ == "__main__":
+    main()

tests/test_fetch_hosted_inputs.py

@@ -0,0 +1,123 @@
+"""Synthetic-only checks; no cloud, credentials, or market inputs are accessed."""
+
+import hashlib
+import importlib.util
+import json
+from pathlib import Path
+import subprocess
+import tempfile
+import unittest
+from unittest.mock import patch
+
+
+SPEC = importlib.util.spec_from_file_location(
+    "fetch_hosted_inputs", Path(__file__).parents[1] / "scripts/fetch_hosted_inputs.py",
+)
+fetcher = importlib.util.module_from_spec(SPEC)
+SPEC.loader.exec_module(fetcher)
+
+
+class CloudInputTests(unittest.TestCase):
+    def setUp(self):
+        self.temporary = tempfile.TemporaryDirectory()
+        self.addCleanup(self.temporary.cleanup)
+        self.payload = b"synthetic input, never real market data"
+        self.env = {
+            "HOSTED_INPUT_FETCH_ENABLED": "true", "GITHUB_ACTIONS": "true",
+            "RUNNER_ENVIRONMENT": "github-hosted", "GITHUB_REF": "refs/heads/main",
+            "HOSTED_INPUT_URI": "gs://example-inputs/approved/bundle.zip",
+            "HOSTED_INPUT_GENERATION": "123", "HOSTED_INPUT_ASSET": "fixture.zip",
+            "HOSTED_INPUT_SHA256": hashlib.sha256(self.payload).hexdigest(),
+            "RUNNER_TEMP": self.temporary.name,
+        }
+        self.archive = Path(self.temporary.name) / "approved-input/fixture.zip"
+
+    def fake_cloud(self, arguments, **overrides):
+        if arguments[0] == "objects":
+            return json.dumps({"size": len(self.payload), "generation": "123", **overrides})
+        Path(arguments[2]).write_bytes(self.payload)
+        return ""
+
+    def test_disabled_does_nothing(self):
+        for value in (None, "false", "TRUE", ""):
+            with self.subTest(value=value), patch.object(fetcher, "_gcloud") as cloud:
+                env = {} if value is None else {"HOSTED_INPUT_FETCH_ENABLED": value}
+                self.assertFalse(fetcher.fetch(env))
+                cloud.assert_not_called()
+
+    def test_rejects_non_hosted_or_non_main_execution_before_cloud(self):
+        for key, value in (("GITHUB_ACTIONS", "false"), ("RUNNER_ENVIRONMENT", "self-hosted"),
+                           ("GITHUB_REF", "refs/pull/1/merge")):
+            with self.subTest(key=key), patch.object(fetcher, "_gcloud") as cloud:
+                with self.assertRaises(ValueError):
+                    fetcher.fetch({**self.env, key: value})
+                cloud.assert_not_called()
+
+    def test_requires_exact_object_generation_digest_and_safe_destination(self):
+        cases = [("HOSTED_INPUT_URI", "gs://example-inputs/approved/*"),
+                 ("HOSTED_INPUT_URI", "https://example.invalid/input"),
+                 ("HOSTED_INPUT_URI", "gs://example-inputs/input.zip?token=synthetic"),
+                 ("HOSTED_INPUT_GENERATION", ""), ("HOSTED_INPUT_GENERATION", "0"),
+                 ("HOSTED_INPUT_SHA256", ""), ("HOSTED_INPUT_ASSET", "../bad.zip"),
+                 ("RUNNER_TEMP", "relative")]
+        for key, value in cases:
+            with self.subTest(key=key, value=value), patch.object(fetcher, "_gcloud") as cloud:
+                with self.assertRaises(ValueError):
+                    fetcher.fetch({**self.env, key: value})
+                cloud.assert_not_called()
+
+    def test_exact_generation_get_and_verified_archive(self):
+        with patch.object(fetcher, "_gcloud", side_effect=self.fake_cloud) as cloud:
+            self.assertTrue(fetcher.fetch(self.env))
+        self.assertEqual(self.archive.read_bytes(), self.payload)
+        self.assertEqual(cloud.call_count, 2)
+        for call in cloud.call_args_list:
+            self.assertEqual(call.args[0][2 if call.args[0][0] == "objects" else 1],
+                             "gs://example-inputs/approved/bundle.zip#123")
+            self.assertNotIn("ls", call.args[0])
+
+    def test_rejects_metadata_before_download(self):
+        for overrides in ({"generation": "124"}, {"size": 0},
+                          {"size": fetcher.MAX_ARCHIVE_BYTES + 1}):
+            with self.subTest(overrides=overrides), patch.object(
+                fetcher, "_gcloud", side_effect=lambda args: self.fake_cloud(args, **overrides),
+            ) as cloud:
+                with self.assertRaises(ValueError):
+                    fetcher.fetch(self.env)
+                self.assertEqual(cloud.call_count, 1)
+                self.assertFalse(self.archive.exists())
+
+    def test_digest_or_size_mismatch_never_publishes_partial(self):
+        for mismatch in ("digest", "size"):
+            with self.subTest(mismatch=mismatch):
+                def fake(arguments):
+                    if arguments[0] == "objects":
+                        return json.dumps({"size": len(self.payload) + (mismatch == "size"), "generation": "123"})
+                    return self.fake_cloud(arguments)
+                env = {**self.env, "HOSTED_INPUT_SHA256": "0" * 64} if mismatch == "digest" else self.env
+                with patch.object(fetcher, "_gcloud", side_effect=fake):
+                    with self.assertRaises(ValueError):
+                        fetcher.fetch(env)
+                self.assertFalse(self.archive.exists())
+                self.assertEqual(list(self.archive.parent.iterdir()), [])
+
+    def test_cloud_failure_has_no_fallback_or_private_error_text(self):
+        result = subprocess.CompletedProcess([], 1, "", "private object and authentication details")
+        with patch.object(fetcher.subprocess, "run", return_value=result) as run:
+            with self.assertRaisesRegex(ValueError, r"^approved cloud input command failed \(exit 1\)$"):
+                fetcher.fetch(self.env)
+            self.assertEqual(run.call_count, 1)
+        self.assertFalse(self.archive.exists())
+
+    def test_existing_destination_is_not_overwritten(self):
+        self.archive.parent.mkdir()
+        self.archive.write_bytes(b"prior verified input")
+        with patch.object(fetcher, "_gcloud", side_effect=self.fake_cloud) as cloud:
+            with self.assertRaises(ValueError):
+                fetcher.fetch(self.env)
+            self.assertEqual(cloud.call_count, 1)
+        self.assertEqual(self.archive.read_bytes(), b"prior verified input")
+
+
+if __name__ == "__main__":
+    unittest.main()