diff --git a/.github/workflows/research_input_readback.yml b/.github/workflows/research_input_readback.yml index 5e461f3c..7063911e 100644 --- a/.github/workflows/research_input_readback.yml +++ b/.github/workflows/research_input_readback.yml @@ -14,7 +14,7 @@ jobs: readback: if: github.repository == 'QuantStrategyLab/AIAuditBridge' && github.ref == 'refs/heads/main' runs-on: [self-hosted, codex-vps] - timeout-minutes: 5 + timeout-minutes: 10 permissions: contents: read id-token: write @@ -24,6 +24,26 @@ jobs: with: persist-credentials: false + - name: Checkout the frozen original P1 validator + uses: actions/checkout@df4cb1c069e1874edd31b4311f1884172cec0e10 # v6.0.3 + with: + repository: QuantStrategyLab/UsEquitySnapshotPipelines + ref: ca61b82c2a508a1cc81fb5831294ba9835ac41c2 + path: validator-source + persist-credentials: false + + - uses: actions/setup-python@a26af69be951a213d495a4c3e4e4022e16d87065 # v6.2.0 + with: + python-version: '3.11' + + # Match the source workflow's locked environment before minting the + # short-lived cloud credential, so installation cannot consume its life. + - name: Install the frozen validator runtime + working-directory: validator-source + run: | + python -m pip install --quiet 'uv==0.11.19' + env -u UV_PYTHON -u VIRTUAL_ENV uv sync --locked --no-dev --no-editable --python 3.11 + # Direct federation grants only object reads on the fixed P1 root below. # The provider additionally binds numeric repo/owner IDs, main and this workflow. # No service-account key, user ADC, daemon credential or AI invocation is used. @@ -36,16 +56,33 @@ jobs: export_environment_variables: true cleanup_credentials: true - - name: Read completion marker metadata without downloading research data + - name: Read and verify the completed immutable P1 root env: CLOUDSDK_CORE_DISABLE_PROMPTS: '1' CLOUDSDK_CORE_DISABLE_USAGE_REPORTING: 'true' + INPUT_ROOT: gs://qsl-runtime-logs-shared/soxl-p1-p3/06ce97ad581fdc465896ef324ce16ac65f2695af95ec18ce27238bdd71032f74/ + EXPECTED_MANIFEST_SHA256: 06ce97ad581fdc465896ef324ce16ac65f2695af95ec18ce27238bdd71032f74 + READBACK_ROOT: ${{ runner.temp }}/aab-research-input-${{ github.run_id }}-${{ github.run_attempt }} + VALIDATOR_SOURCE: ${{ github.workspace }}/validator-source run: | + set -euo pipefail + cleanup() { + rm -rf -- "$READBACK_ROOT" + } + trap cleanup EXIT + if [ -e "$READBACK_ROOT" ] || [ -L "$READBACK_ROOT" ]; then + echo '{"status":"unavailable","reason":"temporary_root_unavailable"}' + exit 1 + fi + umask 077 + mkdir -m 700 "$READBACK_ROOT" + mkdir -m 700 "$READBACK_ROOT/root" + python3 - <<'PY' import json import os + import stat import subprocess - from datetime import datetime from pathlib import Path def fail(reason): @@ -66,27 +103,87 @@ jobs: except (OSError, ValueError, KeyError): fail('temporary_identity_invalid') - root = 'gs://qsl-runtime-logs-shared/soxl-p1-p3/06ce97ad581fdc465896ef324ce16ac65f2695af95ec18ce27238bdd71032f74/' + source = os.environ['INPUT_ROOT'] + workspace = Path(os.environ['READBACK_ROOT']) + local_root = workspace / 'root' + members = { + 'binding.json': (local_root / 'binding.json', 1024 * 1024), + 'bars.json': (local_root / 'bars.json', 32 * 1024 * 1024), + 'manifest.json': (local_root / 'manifest.json', 1024 * 1024), + 'p1-complete.json': (workspace / 'p1-complete.json', 1024 * 1024), + } + observed_sizes = {} + for name, (_, maximum) in members.items(): + try: + result = subprocess.run( + ['gcloud', 'storage', 'objects', 'describe', source + name, '--raw', '--format=json'], + capture_output=True, text=True, timeout=30, check=False, + ) + except (OSError, subprocess.TimeoutExpired): + fail('metadata_read_unavailable') + if result.returncode: + fail('metadata_read_unavailable') + try: + size = int(json.loads(result.stdout)['size']) + except (KeyError, ValueError, TypeError, json.JSONDecodeError): + fail('metadata_invalid') + if size <= 0 or size > maximum: + fail('object_size_invalid') + observed_sizes[name] = size + if sum(observed_sizes.values()) > 35 * 1024 * 1024: + fail('read_budget_exceeded') + + for name, (destination, _) in members.items(): + try: + result = subprocess.run( + ['gcloud', 'storage', 'cp', '--quiet', source + name, str(destination)], + capture_output=True, text=True, timeout=60, check=False, + ) + except (OSError, subprocess.TimeoutExpired): + fail('object_read_unavailable') + if result.returncode: + fail('object_read_unavailable') + try: + item_stat = destination.lstat() + except OSError: + fail('object_read_unavailable') + if not stat.S_ISREG(item_stat.st_mode) or item_stat.st_size != observed_sizes[name]: + fail('object_read_invalid') + PY + + cd "$VALIDATOR_SOURCE" + ROOT="$READBACK_ROOT/root" COMPLETION="$READBACK_ROOT/p1-complete.json" \ + env -u UV_PYTHON -u VIRTUAL_ENV uv run --no-sync python - <<'PY' + import json + import os + from pathlib import Path + try: - result = subprocess.run( - ['gcloud', 'storage', 'objects', 'describe', root + 'p1-complete.json', '--raw', '--format=json'], - capture_output=True, text=True, timeout=30, check=False, + from us_equity_snapshot_pipelines.lifecycle.soxl_core_only_p1_publisher import ( + verify_soxl_core_only_input_root, + verify_soxl_core_only_p1_remote_completion, ) - except (OSError, subprocess.TimeoutExpired): - fail('metadata_read_unavailable') - if result.returncode: - fail('metadata_read_unavailable') - try: - metadata = json.loads(result.stdout) - size = int(metadata['size']) - created_at = datetime.fromisoformat(metadata['timeCreated'].replace('Z', '+00:00')) - if size <= 0 or created_at.utcoffset() is None: - fail('metadata_invalid') - except (KeyError, ValueError, TypeError, AttributeError): - fail('metadata_invalid') + manifest_sha256 = verify_soxl_core_only_input_root(Path(os.environ['ROOT'])) + if manifest_sha256 != os.environ['EXPECTED_MANIFEST_SHA256']: + raise ValueError + if verify_soxl_core_only_p1_remote_completion( + Path(os.environ['ROOT']), Path(os.environ['COMPLETION']) + ) != manifest_sha256: + raise ValueError + except Exception: + print(json.dumps({'status': 'unavailable', 'reason': 'p1_validation_failed'})) + raise SystemExit(1) from None print(json.dumps({ - 'status': 'available', 'identity': 'github_oidc', - 'size': size, 'created_at': created_at.isoformat(), - 'metadata_only': True, 'research_executed': False, + 'status': 'accepted', + 'identity': 'github_oidc', + 'manifest_sha256': manifest_sha256, + 'member_count': 4, + 'research_executed': False, })) PY + + - name: Remove the bounded readback workspace + if: always() + env: + READBACK_ROOT: ${{ runner.temp }}/aab-research-input-${{ github.run_id }}-${{ github.run_attempt }} + run: rm -rf -- "$READBACK_ROOT" diff --git a/tests/test_research_input_readback_workflow.py b/tests/test_research_input_readback_workflow.py new file mode 100644 index 00000000..5c770d66 --- /dev/null +++ b/tests/test_research_input_readback_workflow.py @@ -0,0 +1,214 @@ +"""Behavior checks for the bounded VPS P1 input readback workflow.""" + +from __future__ import annotations + +import json +import subprocess +import sys +import textwrap +import types +from pathlib import Path +from types import SimpleNamespace + +import pytest + +WORKFLOW = Path(__file__).resolve().parents[1] / ".github/workflows/research_input_readback.yml" +OBJECTS = { + "binding.json": b"binding", + "bars.json": b"bars-data", + "manifest.json": b"manifest", + "p1-complete.json": b"completion", +} + + +def workflow_text() -> str: + return WORKFLOW.read_text() + + +def python_blocks() -> tuple[str, str]: + parts = workflow_text().split("<<'PY'\n")[1:] + assert len(parts) == 2 + blocks = [textwrap.dedent(part.split("\n PY", maxsplit=1)[0]) for part in parts] + return blocks[0], blocks[1] + + +def configure_readback_environment(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> Path: + credential = tmp_path / "gha-creds.json" + credential.write_text('{"type":"external_account"}') + workspace = tmp_path / "readback" + (workspace / "root").mkdir(parents=True, mode=0o700) + monkeypatch.setenv("GOOGLE_APPLICATION_CREDENTIALS", str(credential)) + monkeypatch.setenv("CLOUDSDK_AUTH_CREDENTIAL_FILE_OVERRIDE", str(credential)) + monkeypatch.setenv("INPUT_ROOT", "gs://fixed-root/") + monkeypatch.setenv("READBACK_ROOT", str(workspace)) + return workspace + + +def install_fake_validator( + monkeypatch: pytest.MonkeyPatch, *, manifest_sha256: str, reject: bool = False +) -> list[tuple[str, Path, Path | None]]: + calls: list[tuple[str, Path, Path | None]] = [] + package = types.ModuleType("us_equity_snapshot_pipelines") + lifecycle = types.ModuleType("us_equity_snapshot_pipelines.lifecycle") + publisher = types.ModuleType( + "us_equity_snapshot_pipelines.lifecycle.soxl_core_only_p1_publisher" + ) + + def verify_root(root: Path) -> str: + calls.append(("root", root, None)) + if reject: + raise ValueError("hidden input detail") + return manifest_sha256 + + def verify_completion(root: Path, marker: Path) -> str: + calls.append(("completion", root, marker)) + return manifest_sha256 + + publisher.verify_soxl_core_only_input_root = verify_root # type: ignore[attr-defined] + publisher.verify_soxl_core_only_p1_remote_completion = verify_completion # type: ignore[attr-defined] + monkeypatch.setitem(sys.modules, package.__name__, package) + monkeypatch.setitem(sys.modules, lifecycle.__name__, lifecycle) + monkeypatch.setitem(sys.modules, publisher.__name__, publisher) + return calls + + +def test_success_reads_exactly_four_objects_and_calls_both_original_gates( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path, capsys: pytest.CaptureFixture[str] +) -> None: + download_block, validator_block = python_blocks() + workspace = configure_readback_environment(monkeypatch, tmp_path) + subprocess_calls: list[list[str]] = [] + + def fake_run(args: list[str], **_kwargs: object) -> SimpleNamespace: + subprocess_calls.append(args) + name = args[4].removeprefix("gs://fixed-root/") + if args[2:4] == ["objects", "describe"]: + return SimpleNamespace(returncode=0, stdout=json.dumps({"size": len(OBJECTS[name])})) + destination = Path(args[-1]) + destination.write_bytes(OBJECTS[name]) + return SimpleNamespace(returncode=0, stdout="") + + monkeypatch.setattr(subprocess, "run", fake_run) + exec(compile(download_block, "", "exec"), {}) + + manifest_sha256 = "a" * 64 + monkeypatch.setenv("ROOT", str(workspace / "root")) + monkeypatch.setenv("COMPLETION", str(workspace / "p1-complete.json")) + monkeypatch.setenv("EXPECTED_MANIFEST_SHA256", manifest_sha256) + validator_calls = install_fake_validator(monkeypatch, manifest_sha256=manifest_sha256) + exec(compile(validator_block, "", "exec"), {}) + + described = [call[4] for call in subprocess_calls if call[2:4] == ["objects", "describe"]] + copied = [call[4] for call in subprocess_calls if call[2:4] == ["cp", "--quiet"]] + assert described == ["gs://fixed-root/" + name for name in OBJECTS] + assert copied == ["gs://fixed-root/" + name for name in OBJECTS] + assert [call[0] for call in validator_calls] == ["root", "completion"] + assert json.loads(capsys.readouterr().out) == { + "status": "accepted", + "identity": "github_oidc", + "manifest_sha256": manifest_sha256, + "member_count": 4, + "research_executed": False, + } + + +def test_oversize_object_is_rejected_before_any_copy( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path, capsys: pytest.CaptureFixture[str] +) -> None: + download_block, _ = python_blocks() + configure_readback_environment(monkeypatch, tmp_path) + calls: list[list[str]] = [] + + def fake_run(args: list[str], **_kwargs: object) -> SimpleNamespace: + calls.append(args) + return SimpleNamespace(returncode=0, stdout=json.dumps({"size": 1024 * 1024 + 1})) + + monkeypatch.setattr(subprocess, "run", fake_run) + with pytest.raises(SystemExit): + exec(compile(download_block, "", "exec"), {}) + + assert not any(call[2:4] == ["cp", "--quiet"] for call in calls) + assert json.loads(capsys.readouterr().out) == { + "status": "unavailable", + "reason": "object_size_invalid", + } + + +def test_missing_temporary_identity_never_calls_gcloud( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path, capsys: pytest.CaptureFixture[str] +) -> None: + download_block, _ = python_blocks() + workspace = tmp_path / "readback" + (workspace / "root").mkdir(parents=True, mode=0o700) + monkeypatch.setenv("READBACK_ROOT", str(workspace)) + monkeypatch.delenv("GOOGLE_APPLICATION_CREDENTIALS", raising=False) + monkeypatch.delenv("CLOUDSDK_AUTH_CREDENTIAL_FILE_OVERRIDE", raising=False) + + def forbidden_run(*_args: object, **_kwargs: object) -> None: + raise AssertionError("gcloud must not run") + + monkeypatch.setattr(subprocess, "run", forbidden_run) + with pytest.raises(SystemExit): + exec(compile(download_block, "", "exec"), {}) + + assert json.loads(capsys.readouterr().out) == { + "status": "unavailable", + "reason": "temporary_identity_missing", + } + + +def test_copy_failure_outputs_only_the_fixed_safe_category( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path, capsys: pytest.CaptureFixture[str] +) -> None: + download_block, _ = python_blocks() + configure_readback_environment(monkeypatch, tmp_path) + + def fake_run(args: list[str], **_kwargs: object) -> SimpleNamespace: + if args[2:4] == ["objects", "describe"]: + name = args[4].removeprefix("gs://fixed-root/") + return SimpleNamespace(returncode=0, stdout=json.dumps({"size": len(OBJECTS[name])})) + return SimpleNamespace(returncode=1, stdout="", stderr="sensitive provider detail") + + monkeypatch.setattr(subprocess, "run", fake_run) + with pytest.raises(SystemExit): + exec(compile(download_block, "", "exec"), {}) + + output = capsys.readouterr().out + assert json.loads(output) == {"status": "unavailable", "reason": "object_read_unavailable"} + assert "sensitive" not in output + assert "accepted" not in output + + +def test_validator_rejection_outputs_only_the_fixed_safe_category( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path, capsys: pytest.CaptureFixture[str] +) -> None: + _, validator_block = python_blocks() + manifest_sha256 = "a" * 64 + monkeypatch.setenv("ROOT", str(tmp_path / "root")) + monkeypatch.setenv("COMPLETION", str(tmp_path / "p1-complete.json")) + monkeypatch.setenv("EXPECTED_MANIFEST_SHA256", manifest_sha256) + install_fake_validator(monkeypatch, manifest_sha256=manifest_sha256, reject=True) + + with pytest.raises(SystemExit): + exec(compile(validator_block, "", "exec"), {}) + + output = capsys.readouterr().out + assert json.loads(output) == {"status": "unavailable", "reason": "p1_validation_failed"} + assert "hidden" not in output + assert "accepted" not in output + + +def test_workflow_keeps_frozen_source_main_identity_and_no_research_execution() -> None: + text = workflow_text() + + assert "github.ref == 'refs/heads/main'" in text + assert "repository: QuantStrategyLab/UsEquitySnapshotPipelines" in text + assert "ref: ca61b82c2a508a1cc81fb5831294ba9835ac41c2" in text + assert "uv sync --locked --no-dev --no-editable --python 3.11" in text + assert text.index("name: Install the frozen validator runtime") < text.index( + "name: Authenticate for this research input only" + ) + assert "trap cleanup EXIT" in text + assert "if: always()" in text + assert "upload-artifact" not in text + assert "run_soxl_core_only_p3_evidence" not in text