From 97f410ee22e6db63814c3c1ff79d92ad67287781 Mon Sep 17 00:00:00 2001 From: David Spencer <1526975+DecisionNerd@users.noreply.github.com> Date: Sun, 30 Aug 2026 20:27:37 -0600 Subject: [PATCH 1/2] build(bench): add offline provider qualification runner --- .github/workflows/test.yml | 24 + benchmarks/Makefile | 3 +- benchmarks/README.md | 30 +- .../progressive_provider_plan.py | 61 +- .../progressive_provider_run.py | 489 +++++++++++++++ .../graphforge_bench/progressive_run.py | 51 +- .../qualification_operator.py | 8 +- .../schemas/progressive-provider-plan.json | 8 +- .../progressive-provider-run-plan.json | 50 ++ .../progressive-provider-run-result.json | 42 +- .../tests/test_progressive_provider_plan.py | 117 +++- .../tests/test_progressive_provider_run.py | 584 ++++++++++++++++++ benchmarks/tests/test_progressive_run.py | 44 ++ .../tests/test_qualification_operator.py | 2 +- .../Dockerfile | 73 +++ .../run-qualification.py | 82 +++ .../test-progressive-qualification-image.py | 451 ++++++++++++++ 17 files changed, 2075 insertions(+), 44 deletions(-) create mode 100644 benchmarks/harness/graphforge_bench/progressive_provider_run.py create mode 100644 benchmarks/schemas/progressive-provider-run-plan.json create mode 100644 benchmarks/tests/test_progressive_provider_run.py create mode 100644 containers/graphforge-progressive-qualification/Dockerfile create mode 100644 containers/graphforge-progressive-qualification/run-qualification.py create mode 100644 scripts/ci/test-progressive-qualification-image.py diff --git a/.github/workflows/test.yml b/.github/workflows/test.yml index b5888b94..33cd5968 100644 --- a/.github/workflows/test.yml +++ b/.github/workflows/test.yml @@ -85,6 +85,9 @@ jobs: - name: Test Fly qualification safety contract run: python3 scripts/ci/test-fly-filesystem-safety-contract.py + - name: Test progressive qualification image contract + run: python3 scripts/ci/test-progressive-qualification-image.py + - name: Test Repository Policy suite classifier run: scripts/ci/test-classify-policy-suites.sh @@ -1169,6 +1172,27 @@ jobs: with: enable-cache: true + - name: Test offline progressive provider boundary + working-directory: benchmarks + run: | + PYTHONPATH=harness uv run --locked python -m unittest \ + tests.test_progressive_provider_plan \ + tests.test_progressive_provider_run \ + tests.test_progressive_run \ + tests.test_qualification_operator + + - name: Build immutable progressive qualification image + shell: bash + run: | + set -euo pipefail + source_sha=$(git rev-parse HEAD) + image="graphforge-progressive-qualification:${source_sha}" + docker build --platform linux/amd64 \ + --build-arg "GRAPHFORGE_COMMIT=${source_sha}" \ + --file containers/graphforge-progressive-qualification/Dockerfile \ + --tag "$image" . + test "$(docker image inspect --format '{{ index .Config.Labels "org.opencontainers.image.revision" }}' "$image")" = "$source_sha" + - name: Cargo feature drift check (fail-closed) run: | python3 scripts/ci/cargo-bazel-drift-check.py diff --git a/benchmarks/Makefile b/benchmarks/Makefile index 685c3e8e..549b2710 100644 --- a/benchmarks/Makefile +++ b/benchmarks/Makefile @@ -54,11 +54,12 @@ progressive-qualification-project-s20: install # Safe control-plane admission only; this never provisions a provider. progressive-provider-plan: install - test -n "$(COMMIT)" && test -n "$(MAXIMUM_SCALE)" && test -n "$(OUTPUT_DIR)" && test -n "$(PLAN_OUT)" + test -n "$(COMMIT)" && test -n "$(MAXIMUM_SCALE)" && test -n "$(OUTPUT_DIR)" && test -n "$(PLAN_OUT)" && test -n "$(IMAGE_DIGEST)" PYTHONPATH=$(CURDIR)/harness uv run --locked python -m graphforge_bench.qualification_operator \ plan-progressive -- \ --root "$(CURDIR)" --output-dir "$(OUTPUT_DIR)" --commit "$(COMMIT)" \ --maximum-scale "$(MAXIMUM_SCALE)" $(if $(PROVIDER_CAPACITY),--provider-capacity "$(PROVIDER_CAPACITY)",) \ + --image-digest "$(IMAGE_DIGEST)" \ --plan-out "$(PLAN_OUT)" # The registry owns the live command. Pulumi ESC supplies provider credentials; diff --git a/benchmarks/README.md b/benchmarks/README.md index aac36787..a91f8ea2 100644 --- a/benchmarks/README.md +++ b/benchmarks/README.md @@ -307,14 +307,26 @@ make -C benchmarks progressive-provider-plan \ COMMIT=$(git rev-parse HEAD) MAXIMUM_SCALE=26 \ OUTPUT_DIR=/admitted-volume/graphforge-evidence \ PLAN_OUT=/admitted-volume/graphforge-evidence/provider-plan.json \ + IMAGE_DIGEST=registry.fly.io/graphforge-bench@sha256:<64-hex-digest> \ PROVIDER_CAPACITY=/sanitized/provider-capacity.json ``` -The planner is deliberately not a provider executor. Provider plans are marked -refused for execution until a dedicated provider image contains `gf`, the -Graph500 generator/certifier, Python/BenchExec, and a proven cgroups/I/O -authority. The canonical order remains S18, S19, S20, S22, S24, S25, S26; -the first failed or missing gate stops the ladder. +An admitted S20--S26 plan can be consumed by the offline provider runner inside +the dedicated `containers/graphforge-progressive-qualification` image. The +runner requires the image's read-only build manifest, matches the immutable +`registry.fly.io/...@sha256:...` identity supplied by the admitted plan and +provider transport, and revalidates the canonical profile, projection, source +tree, native executables, and BenchExec identity before starting BenchExec. It +accepts only fixed in-image executable paths, an admitted plan and evidence +directory below `/work`, no pre-existing files for the selected rung, and a +real `/work` mount. The host +must separately pass native Linux cgroups-v2 admission. The runner has no +laptop fallback and never calls a provider API. + +A successful rung emits exactly `sN-plan.json`, `sN-benchexec.json`, +`sN-graphforge.json`, `sN-rung.json`, and `sN-result.json`, all scoped to +engineering evidence. The canonical order remains S18, S19, S20, S22, S24, +S25, S26; the first failed or missing gate stops the planner. Provider credentials belong to Pulumi ESC rather than GitHub workflow inputs or the caller's ambient shell. Live operator commands are rendered from @@ -329,9 +341,11 @@ make -C benchmarks qualification-operator \ The operator uses the shell-free form `pulumi env run -- `; secret values are never copied into its command line or evidence. The -`progressive-ladder` command is registered now, but fails before opening ESC -until the dedicated provider image and scale BenchExec executor are present. -This is a deliberate capability boundary, not a GitHub-dispatch prerequisite. +`progressive-ladder` is registered, but still fails before opening ESC. The +offline image and rung runner do not implement whole-attempt Fly orchestration, +typed spend authorization, ownership-ledger recovery, or teardown inventory. +This is a deliberate live capability boundary, not a GitHub-dispatch +prerequisite. The controller derives bulk-ingest capability from the same run's bounded ordinary `gf import-session commit --json` receipt: its construction evidence diff --git a/benchmarks/harness/graphforge_bench/progressive_provider_plan.py b/benchmarks/harness/graphforge_bench/progressive_provider_plan.py index 7b8a1d3b..e0702a05 100644 --- a/benchmarks/harness/graphforge_bench/progressive_provider_plan.py +++ b/benchmarks/harness/graphforge_bench/progressive_provider_plan.py @@ -29,8 +29,8 @@ SCALES = (18, 19, 20, 22, 24, 25, 26) COMMIT = re.compile(r"^[0-9a-f]{40}$") +IMAGE_DIGEST = re.compile(r"^registry\.fly\.io/[a-z0-9][a-z0-9._/-]*@sha256:[0-9a-f]{64}$") PLAN_SCHEMA = "graphforge-progressive-provider-plan/1" -EXECUTION_REFUSAL = "provider_executor_unavailable" class ProviderPlanError(ValueError): @@ -87,14 +87,14 @@ def _validate_rung(root: Path, value: Any, scale: int) -> Mapping[str, Any]: error = next(Draft202012Validator(schema).iter_errors(value), None) if error is not None: raise ProviderPlanError("completed rung evidence is not schema-valid") - expected_source = "progressive_profile" if scale in (18, 19) else None + expected_source = "progressive_profile" if scale in (18, 19) else "canonical_ladder" if ( value.get("scale") != scale or value.get("status") != "passed" or value.get("live_edges") != 16 * (1 << scale) or value.get("profile_id") != f"graph500-s{scale}-{'local' if scale in (18, 19) else 'provider'}" - or (expected_source is not None and value.get("source") != expected_source) + or value.get("source") != expected_source ): raise ProviderPlanError("completed rung evidence does not match its canonical profile") return value @@ -135,6 +135,40 @@ def _validate_result_identity( raise ProviderPlanError( "completed rung result is not bound to the requested commit/profile" ) + if scale not in (18, 19): + artifacts = value.get("artifacts") + execution_plan = _read_json( + output_dir / f"s{scale}-plan.json", + "completed provider execution plan is unavailable", + ) + execution_schema = _read_json( + root / "schemas" / "progressive-provider-run-plan.json", + "provider execution plan schema is unavailable", + ) + execution_error = next( + Draft202012Validator(execution_schema).iter_errors(execution_plan), None + ) + if ( + execution_error is not None + or not isinstance(execution_plan, Mapping) + or execution_plan.get("rung") != f"S{scale}" + or execution_plan.get("identities") != identities + ): + raise ProviderPlanError("completed provider plan/result identities disagree") + try: + expected_artifacts = { + name: hashlib.sha256(path.read_bytes()).hexdigest() + for name, path in { + "plan_sha256": output_dir / f"s{scale}-plan.json", + "benchexec_sha256": output_dir / f"s{scale}-benchexec.json", + "graphforge_sha256": output_dir / f"s{scale}-graphforge.json", + "rung_sha256": output_dir / f"s{scale}-rung.json", + }.items() + } + except OSError as error: + raise ProviderPlanError("completed provider artifacts are unavailable") from error + if not isinstance(artifacts, Mapping) or dict(artifacts) != expected_artifacts: + raise ProviderPlanError("completed provider artifacts do not match their result") def completed_rungs( @@ -177,6 +211,7 @@ def plan_provider_ladder( commit: str, maximum_scale: int, provider_capacity: Mapping[str, Any] | None = None, + image_digest: str | None = None, ) -> dict[str, Any]: """Return one immutable, sanitized next-rung plan without provider calls.""" if COMMIT.fullmatch(commit) is None: @@ -198,6 +233,8 @@ def plan_provider_ladder( profile_path = _profile_path(root, selected) projection: Mapping[str, Any] | None = None if selected.execution == "provider": + if image_digest is None or IMAGE_DIGEST.fullmatch(image_digest) is None: + raise ProviderPlanError("immutable provider image digest is required") try: projection = project(selected, completed, provider_capacity) except QualificationError as error: @@ -222,9 +259,10 @@ def plan_provider_ladder( "profile_id": selected.id, "profile_path": profile_path.relative_to(root).as_posix(), "profile_sha256": "sha256:" + _sha256(profile_path), + "image_digest": image_digest if selected.execution == "provider" else None, "projection": projection, - "execution_authorized": selected.execution == "local", - "execution_refusal": None if selected.execution == "local" else EXECUTION_REFUSAL, + "execution_authorized": True, + "execution_refusal": None, "claim": "engineering_evidence_only", } schema_path = root / "schemas" / "progressive-provider-plan.json" @@ -236,22 +274,20 @@ def plan_provider_ladder( def require_execution_authority(plan: Mapping[str, Any]) -> None: - """Refuse provider execution until its image and resource authority exist.""" - if plan.get("execution") == "provider": - raise ProviderPlanError( - "provider execution is refused until a dedicated provider image and BenchExec " - "authority exist" - ) + """Require the checked-in offline runner before handing a plan to execution.""" + if plan.get("execution_authorized") is not True or plan.get("execution_refusal") is not None: + raise ProviderPlanError("provider execution authority is unavailable") def main(argv: list[str] | None = None) -> int: """Write one no-spend provider plan for a protected workflow step.""" - parser = argparse.ArgumentParser(description=__doc__) + parser = argparse.ArgumentParser(description=__doc__, allow_abbrev=False) parser.add_argument("--root", type=Path, required=True) parser.add_argument("--output-dir", type=Path, required=True) parser.add_argument("--commit", required=True) parser.add_argument("--maximum-scale", type=int, required=True) parser.add_argument("--provider-capacity", type=Path) + parser.add_argument("--image-digest") parser.add_argument("--plan-out", type=Path, required=True) args = parser.parse_args(argv) try: @@ -267,6 +303,7 @@ def main(argv: list[str] | None = None) -> int: commit=args.commit, maximum_scale=args.maximum_scale, provider_capacity=capacity, + image_digest=args.image_digest, ) args.plan_out.parent.mkdir(parents=True, exist_ok=True) args.plan_out.write_text( diff --git a/benchmarks/harness/graphforge_bench/progressive_provider_run.py b/benchmarks/harness/graphforge_bench/progressive_provider_run.py new file mode 100644 index 00000000..58b94289 --- /dev/null +++ b/benchmarks/harness/graphforge_bench/progressive_provider_run.py @@ -0,0 +1,489 @@ +"""Offline, fail-closed executor for one admitted provider ladder rung. + +This module runs *inside* an already admitted provider host. It never calls a +provider API. Provisioning and transport are deliberately outside this trust +boundary; the only default process execution is the existing native BenchExec +lifecycle. +""" + +from __future__ import annotations + +import argparse +from collections.abc import Callable, Mapping, Sequence +import hashlib +from importlib.metadata import PackageNotFoundError, version +import json +from pathlib import Path +import re +import tempfile +from typing import Any + +from jsonschema import Draft202012Validator + +from graphforge_bench.progressive_run import ( + ControllerError, + Executables, + _digest, + _native_authority, + _run_benchexec, + _safe_stage, + _write_json, + ingest_benchexec_result, + repository_commit, + resolve_executables, +) + +PLAN_SCHEMA = "graphforge-progressive-provider-execution-plan/1" +RESULT_SCHEMA = "graphforge-progressive-provider-run-result/1" +BUILD_SCHEMA = "graphforge-progressive-provider-build/1" +PROVIDER_RUNGS = (20, 22, 24, 25, 26) +LADDER_RUNGS = (18, 19, 20, 22, 24, 25, 26) +IMAGE_DIGEST = re.compile(r"^registry\.fly\.io/[a-z0-9][a-z0-9._/-]*@sha256:[0-9a-f]{64}$") +HEX_DIGEST = re.compile(r"^[0-9a-f]{64}$") +COMMIT = re.compile(r"^[0-9a-f]{40}$") + +ExecutionBoundary = Callable[[Path, Executables, Mapping[str, Any]], int] +AuthorityBoundary = Callable[[], Mapping[str, Any]] + + +class ProviderRunError(ControllerError): + """The admitted rung or immutable execution boundary is invalid.""" + + +def _execution_commit(root: Path, *, require_image: bool = False) -> str: + """Read the image attestation, falling back to Git for local no-spend tests.""" + attestation = root.parent / "commit" + if not attestation.exists(): + if require_image: + raise ProviderRunError("image commit attestation is unavailable") + return repository_commit(root) + try: + commit = attestation.read_text(encoding="ascii").strip() + except (OSError, UnicodeDecodeError) as error: + raise ProviderRunError("image commit attestation is unavailable") from error + if COMMIT.fullmatch(commit) is None: + raise ProviderRunError("image commit attestation is malformed") + return commit + + +def _read_build_manifest(root: Path, path: Path) -> tuple[Mapping[str, Any], Executables]: + """Load the read-only build identity and fixed production executables.""" + manifest, _ = _read_document(path) + executable_identities = manifest.get("executables") + if ( + manifest.get("schema") != BUILD_SCHEMA + or manifest.get("commit") != _execution_commit(root, require_image=True) + or HEX_DIGEST.fullmatch(str(manifest.get("source_tree_sha256"))) is None + or not isinstance(executable_identities, Mapping) + or set(executable_identities) + != { + "gf_sha256", + "certify_sha256", + "generator_executable_sha256", + "benchexec_python_sha256", + } + or any(HEX_DIGEST.fullmatch(str(value)) is None for value in executable_identities.values()) + ): + raise ProviderRunError("provider build manifest is malformed") + executables = resolve_executables( + gf="/usr/local/bin/gf", + certify="/usr/local/bin/graphforge-benchmark-certify", + generator="/usr/local/bin/graphforge-benchmark-graph500-generator", + benchexec_python="/opt/graphforge/benchmarks/.venv/bin/python", + ) + actual = { + "gf_sha256": _digest(executables.gf), + "certify_sha256": _digest(executables.certify), + "generator_executable_sha256": _digest(executables.generator), + "benchexec_python_sha256": _digest(executables.benchexec_python), + } + if dict(executable_identities) != actual: + raise ProviderRunError("provider build executable identity mismatch") + return manifest, executables + + +def _confined_work_path(path: Path, *, kind: str) -> Path: + """Require production inputs and outputs to stay below the provider volume.""" + work = Path("/work").resolve(strict=True) + try: + resolved = path.resolve(strict=kind == "input") + except OSError as error: + raise ProviderRunError(f"{kind} path is unavailable") from error + if work not in resolved.parents or resolved == work or path.is_symlink(): + raise ProviderRunError(f"{kind} path must be confined below /work") + if kind == "output": + if resolved.exists() and not resolved.is_dir(): + raise ProviderRunError("output path must be a directory") + elif not resolved.is_file(): + raise ProviderRunError("input path is unavailable") + return resolved + + +def _require_fresh_outputs(output_dir: Path, plan: Mapping[str, Any]) -> None: + outputs = plan.get("outputs") + if not isinstance(outputs, list) or any(not isinstance(name, str) for name in outputs): + raise ProviderRunError("provider execution output contract is malformed") + if any((output_dir / name).exists() or (output_dir / name).is_symlink() for name in outputs): + raise ProviderRunError("selected rung already has evidence") + + +def _require_work_mount() -> None: + work = Path("/work") + if not work.is_dir() or work.is_symlink() or work.resolve() != work or not work.is_mount(): + raise ProviderRunError("provider work volume is unavailable") + + +def _read_document(path: Path) -> tuple[Mapping[str, Any], str]: + try: + encoded = path.read_bytes() + value = json.loads(encoded) + except (OSError, UnicodeDecodeError, json.JSONDecodeError) as error: + raise ProviderRunError("required JSON document is unavailable or malformed") from error + if not isinstance(value, Mapping): + raise ProviderRunError("required JSON document must be an object") + return value, hashlib.sha256(encoded).hexdigest() + + +def _schema(root: Path, name: str, value: Any) -> None: + try: + schema = json.loads((root / "schemas" / name).read_text(encoding="utf-8")) + except (OSError, UnicodeDecodeError, json.JSONDecodeError) as error: + raise ProviderRunError(f"closed schema unavailable: {name}") from error + error = next(Draft202012Validator(schema).iter_errors(value), None) + if error is not None: + raise ProviderRunError(f"{name} validation failed: {error.message}") + + +def _scale(value: Any) -> int: + if not isinstance(value, str) or not re.fullmatch(r"S(20|22|24|25|26)", value): + raise ProviderRunError("admitted plan does not select one provider rung") + return int(value[1:]) + + +def _provider_profile(root: Path, scale: int) -> tuple[Path, Mapping[str, Any]]: + if scale not in PROVIDER_RUNGS: + raise ProviderRunError("provider executor accepts only S20 through S26") + path = root / "profiles" / "graph500" / f"s{scale}-provider.json" + try: + value = json.loads(path.read_text(encoding="utf-8")) + except (OSError, UnicodeDecodeError, json.JSONDecodeError) as error: + raise ProviderRunError("canonical provider profile is unavailable") from error + _schema(root, "progressive-qualification-profile.json", value) + if ( + not isinstance(value, Mapping) + or value.get("id") != f"graph500-s{scale}-provider" + or value.get("scale") != scale + or value.get("execution") != "provider" + ): + raise ProviderRunError("canonical provider profile contradicts the selected rung") + return path, value + + +def validate_admitted_plan( + root: Path, plan: Mapping[str, Any] +) -> tuple[int, Path, Mapping[str, Any]]: + """Validate the planner output and bind it to this exact checkout.""" + _schema(root, "progressive-provider-plan.json", plan) + scale = _scale(plan.get("next_rung")) + profile_path, profile = _provider_profile(root, scale) + expected_relative = f"profiles/graph500/s{scale}-provider.json" + expected_profile_digest = "sha256:" + _digest(profile_path) + projection = plan.get("projection") + expected_completed = list(LADDER_RUNGS[: LADDER_RUNGS.index(scale)]) + expected_sources = profile.get("gate", {}).get("projection_source_scales") + if ( + plan.get("status") != "admitted" + or plan.get("execution") != "provider" + or plan.get("execution_authorized") is not True + or plan.get("execution_refusal") is not None + or plan.get("profile_id") != profile["id"] + or plan.get("profile_path") != expected_relative + or plan.get("profile_sha256") != expected_profile_digest + or plan.get("image_digest") is None + or plan.get("commit") != _execution_commit(root) + or plan.get("completed_scales") != expected_completed + or not isinstance(plan.get("maximum_scale"), int) + or int(plan["maximum_scale"]) < scale + or not isinstance(projection, Mapping) + or projection.get("decision") != "admitted" + or projection.get("target") != f"S{scale}" + or projection.get("source_scales") != expected_sources + ): + raise ProviderRunError("admitted provider plan identity or projection mismatch") + _schema(root, "progressive-qualification-evidence.json", projection) + return scale, profile_path, profile + + +def _benchexec_version() -> str: + try: + return version("BenchExec") + except PackageNotFoundError as error: + raise ProviderRunError("BenchExec package identity unavailable") from error + + +def build_execution_plan( + *, + root: Path, + admitted_plan: Mapping[str, Any], + admitted_plan_sha256: str, + image_digest: str, + executables: Executables, + source_tree_sha256: str, +) -> dict[str, Any]: + """Create the immutable plan consumed by the in-host execution boundary.""" + if HEX_DIGEST.fullmatch(admitted_plan_sha256) is None: + raise ProviderRunError("admitted plan digest must be a bare SHA-256 digest") + if IMAGE_DIGEST.fullmatch(image_digest) is None: + raise ProviderRunError("provider image must be an immutable Fly OCI digest") + if HEX_DIGEST.fullmatch(source_tree_sha256) is None: + raise ProviderRunError("source tree digest must be a bare SHA-256 digest") + scale, profile_path, profile = validate_admitted_plan(root, admitted_plan) + if admitted_plan.get("image_digest") != image_digest: + raise ProviderRunError("running image does not match the admitted image digest") + generator_identity = "sha256:" + _digest(root / "runners/graph500-generator/src/main.rs") + if generator_identity != profile.get("generator", {}).get("identity"): + raise ProviderRunError("generator source identity contradicts the provider profile") + identities = { + "commit": admitted_plan["commit"], + "profile_id": profile["id"], + "profile_sha256": _digest(profile_path), + "image_digest": image_digest, + "generator": generator_identity, + "generator_executable_sha256": _digest(executables.generator), + "gf_sha256": _digest(executables.gf), + "certify_sha256": _digest(executables.certify), + "benchexec_python_sha256": _digest(executables.benchexec_python), + "benchexec_version": _benchexec_version(), + "admitted_plan_sha256": admitted_plan_sha256, + "source_tree_sha256": source_tree_sha256, + } + plan = { + "schema": PLAN_SCHEMA, + "rung": f"S{scale}", + "execution": "provider_native_linux_benchexec", + "identities": identities, + "limits": {"wall_seconds": 14_400, "memory_bytes": 4_294_967_296, "cores": 16}, + "outputs": [ + f"s{scale}-plan.json", + f"s{scale}-benchexec.json", + f"s{scale}-graphforge.json", + f"s{scale}-rung.json", + f"s{scale}-result.json", + ], + "claim": "engineering_evidence_only", + } + _schema(root, "progressive-provider-run-plan.json", plan) + return plan + + +def _assert_identities( + root: Path, + plan: Mapping[str, Any], + executables: Executables, + *, + expected_image_digest: str, + expected_admitted_plan_sha256: str, + expected_source_tree_sha256: str, +) -> tuple[int, Path]: + _schema(root, "progressive-provider-run-plan.json", plan) + scale = _scale(plan.get("rung")) + profile_path, profile = _provider_profile(root, scale) + identities = plan.get("identities") + if not isinstance(identities, Mapping): + raise ProviderRunError("provider execution identities are malformed") + actual = { + "commit": _execution_commit(root), + "profile_id": profile["id"], + "profile_sha256": _digest(profile_path), + "generator": "sha256:" + _digest(root / "runners/graph500-generator/src/main.rs"), + "generator_executable_sha256": _digest(executables.generator), + "gf_sha256": _digest(executables.gf), + "certify_sha256": _digest(executables.certify), + "benchexec_python_sha256": _digest(executables.benchexec_python), + "benchexec_version": _benchexec_version(), + "image_digest": expected_image_digest, + "admitted_plan_sha256": expected_admitted_plan_sha256, + "source_tree_sha256": expected_source_tree_sha256, + } + if any(identities.get(name) != value for name, value in actual.items()): + raise ProviderRunError("provider execution identity changed after planning") + expected_outputs = [ + f"s{scale}-plan.json", + f"s{scale}-benchexec.json", + f"s{scale}-graphforge.json", + f"s{scale}-rung.json", + f"s{scale}-result.json", + ] + if plan.get("outputs") != expected_outputs: + raise ProviderRunError("provider execution output contract changed after planning") + return scale, profile_path + + +def _result( + plan: Mapping[str, Any], + status: str, + failure: str | None, + artifacts: Mapping[str, str] | None = None, +) -> dict[str, Any]: + return { + "schema": RESULT_SCHEMA, + "rung": plan["rung"], + "status": status, + "failure": failure, + "identities": plan["identities"], + "artifacts": artifacts, + "claim": "engineering_evidence_only", + } + + +def run( + *, + root: Path, + output_dir: Path, + plan: Mapping[str, Any], + executables: Executables, + expected_image_digest: str, + expected_admitted_plan_sha256: str, + expected_source_tree_sha256: str, + execution_boundary: ExecutionBoundary = _run_benchexec, + authority_boundary: AuthorityBoundary = _native_authority, +) -> None: + """Execute one provider rung without performing any provider operation.""" + identities = plan.get("identities") + if not isinstance(identities, Mapping): + raise ProviderRunError("provider execution identities are malformed") + scale, profile_path = _assert_identities( + root, + plan, + executables, + expected_image_digest=expected_image_digest, + expected_admitted_plan_sha256=expected_admitted_plan_sha256, + expected_source_tree_sha256=expected_source_tree_sha256, + ) + output_dir.mkdir(parents=True, exist_ok=True) + result_path = output_dir / f"s{scale}-result.json" + try: + authority = authority_boundary() + except (ControllerError, OSError) as error: + failed = _result(plan, "failed", "native_authority_unavailable") + _schema(root, "progressive-provider-run-result.json", failed) + _write_json(result_path, failed) + raise ProviderRunError("native BenchExec authority is unavailable") from error + if not isinstance(authority, Mapping) or authority.get("result") != "passed": + failed = _result(plan, "failed", "native_authority_unavailable") + _schema(root, "progressive-provider-run-result.json", failed) + _write_json(result_path, failed) + raise ProviderRunError("native BenchExec authority is unavailable") + try: + temporary_context = tempfile.TemporaryDirectory( + prefix=".gf-provider-authority-", dir=output_dir + ) + except OSError as error: + failed = _result(plan, "failed", "staging_failed") + _schema(root, "progressive-provider-run-result.json", failed) + _write_json(result_path, failed) + raise ProviderRunError("staging_failed") from error + with temporary_context as temporary: + try: + stage = _safe_stage(root, profile_path, executables, identities, Path(temporary)) + except (ControllerError, OSError) as error: + failed = _result(plan, "failed", "staging_failed") + _schema(root, "progressive-provider-run-result.json", failed) + _write_json(result_path, failed) + raise ProviderRunError("staging_failed") from error + try: + execution_status = execution_boundary(stage, executables, identities) + except Exception as error: + failed = _result(plan, "failed", "execution_boundary_failed") + _schema(root, "progressive-provider-run-result.json", failed) + _write_json(result_path, failed) + raise ProviderRunError("execution_boundary_failed") from error + if execution_status != 0: + failed = _result(plan, "failed", "benchexec_failed") + _schema(root, "progressive-provider-run-result.json", failed) + _write_json(result_path, failed) + raise ProviderRunError("benchexec_failed") + try: + benchexec, graphforge, rung = ingest_benchexec_result( + root=root, + stage=stage, + scale=scale, + plan=plan, + profile_id=str(identities["profile_id"]), + source="canonical_ladder", + ) + except (ControllerError, ValueError) as error: + failed = _result(plan, "failed", "ordinary_receipt_missing") + _schema(root, "progressive-provider-run-result.json", failed) + _write_json(result_path, failed) + raise ProviderRunError("ordinary_receipt_missing") from error + try: + _schema(root, "benchexec-run-evidence.json", benchexec) + _schema(root, "certification-evidence.json", graphforge) + _schema(root, "progressive-qualification-rung-evidence.json", rung) + stored_plan, _ = _read_document(output_dir / f"s{scale}-plan.json") + if dict(stored_plan) != dict(plan): + raise ProviderRunError("stored execution plan changed during the run") + except ProviderRunError as error: + failed = _result(plan, "failed", "ordinary_receipt_missing") + _schema(root, "progressive-provider-run-result.json", failed) + _write_json(result_path, failed) + raise ProviderRunError("ordinary_receipt_missing") from error + artifact_paths = { + "benchexec_sha256": output_dir / f"s{scale}-benchexec.json", + "graphforge_sha256": output_dir / f"s{scale}-graphforge.json", + "rung_sha256": output_dir / f"s{scale}-rung.json", + "plan_sha256": output_dir / f"s{scale}-plan.json", + } + _write_json(artifact_paths["benchexec_sha256"], benchexec) + _write_json(artifact_paths["graphforge_sha256"], graphforge) + _write_json(artifact_paths["rung_sha256"], rung) + artifacts = {name: _digest(path) for name, path in artifact_paths.items()} + passed = _result(plan, "passed", None, artifacts) + _schema(root, "progressive-provider-run-result.json", passed) + _write_json(result_path, passed) + + +def main(argv: Sequence[str] | None = None) -> int: + parser = argparse.ArgumentParser(description=__doc__, allow_abbrev=False) + parser.add_argument("--admitted-plan", type=Path, required=True) + parser.add_argument("--output-dir", type=Path, required=True) + parser.add_argument("--image-digest", required=True) + args = parser.parse_args(argv) + root = Path(__file__).resolve().parents[2] + try: + _require_work_mount() + admitted_path = _confined_work_path(args.admitted_plan, kind="input") + output_dir = _confined_work_path(args.output_dir, kind="output") + admitted, admitted_digest = _read_document(admitted_path) + build_manifest, executables = _read_build_manifest( + root, Path("/opt/graphforge/build-manifest.json") + ) + plan = build_execution_plan( + root=root, + admitted_plan=admitted, + admitted_plan_sha256=admitted_digest, + image_digest=args.image_digest, + executables=executables, + source_tree_sha256=str(build_manifest["source_tree_sha256"]), + ) + output_dir.mkdir(parents=True, exist_ok=True) + _require_fresh_outputs(output_dir, plan) + _write_json(output_dir / f"{str(plan['rung']).lower()}-plan.json", plan) + run( + root=root, + output_dir=output_dir, + plan=plan, + executables=executables, + expected_image_digest=args.image_digest, + expected_admitted_plan_sha256=admitted_digest, + expected_source_tree_sha256=str(build_manifest["source_tree_sha256"]), + ) + return 0 + except (OSError, ControllerError) as error: + print(json.dumps({"schema": RESULT_SCHEMA, "status": "failed", "failure": str(error)})) + return 2 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/benchmarks/harness/graphforge_bench/progressive_run.py b/benchmarks/harness/graphforge_bench/progressive_run.py index c339f4ad..1927a1ba 100644 --- a/benchmarks/harness/graphforge_bench/progressive_run.py +++ b/benchmarks/harness/graphforge_bench/progressive_run.py @@ -75,7 +75,9 @@ def _resolve_executable(value: str, expected_name: str) -> Path: located = str(candidate) if candidate.is_absolute() else shutil.which(value) if located is None: raise ControllerError(f"required executable unavailable: {expected_name}") - resolved = Path(located).resolve(strict=True) + # Keep a venv launcher path intact: resolving its symlink escapes the venv + # and makes the base interpreter unable to import the locked BenchExec. + resolved = Path(os.path.abspath(located)) # noqa: PTH100 if not resolved.is_file() or not os.access(resolved, os.X_OK): raise ControllerError(f"required executable is not executable: {expected_name}") return resolved @@ -191,7 +193,24 @@ def build_plan( def _write_json(path: Path, value: Mapping[str, Any]) -> None: - path.write_text(json.dumps(value, indent=2, sort_keys=True) + "\n", encoding="utf-8") + encoded = (json.dumps(value, indent=2, sort_keys=True) + "\n").encode("utf-8") + descriptor, temporary_name = tempfile.mkstemp(prefix=f".{path.name}.", dir=path.parent) + temporary = Path(temporary_name) + try: + with os.fdopen(descriptor, "wb") as stream: + stream.write(encoded) + stream.flush() + os.fsync(stream.fileno()) + os.link(temporary, path) + temporary.unlink() + directory = os.open(path.parent, os.O_RDONLY | os.O_DIRECTORY) + try: + os.fsync(directory) + finally: + os.close(directory) + except BaseException: + temporary.unlink(missing_ok=True) + raise def write_plan(output_dir: Path, plan: Mapping[str, Any]) -> Path: @@ -276,6 +295,7 @@ def _run_benchexec(stage: Path, executables: Executables, identities: Mapping[st "LANG": "C.UTF-8", "LC_ALL": "C.UTF-8", "PATH": f"{stage / 'bin'}:{Path(sys.executable).parent}:/usr/bin:/bin", + "PYTHONPATH": str(Path(__file__).resolve().parents[1]), } command = [ str(executables.benchexec_python), @@ -481,7 +501,13 @@ def _construction_metrics(import_receipt: Mapping[str, Any]) -> dict[str, int]: def assemble_rung_evidence( - *, root: Path, scale: int, graphforge: Mapping[str, Any], benchexec: Mapping[str, Any] + *, + root: Path, + scale: int, + graphforge: Mapping[str, Any], + benchexec: Mapping[str, Any], + profile_id: str | None = None, + source: str = "progressive_profile", ) -> dict[str, Any]: if graphforge.get("status") != "passed" or benchexec.get("outcome") != "passed": raise ControllerError("a failed execution cannot produce passed rung evidence") @@ -531,8 +557,8 @@ def assemble_rung_evidence( raise ControllerError("BenchExec authority is missing") phases = [phase.get("phase") for phase in graphforge.get("phases", [])] rung = { - "profile_id": f"graph500-s{scale}-local", - "source": "progressive_profile", + "profile_id": profile_id or f"graph500-s{scale}-local", + "source": source, "scale": scale, "live_edges": expected_edges, "status": "passed", @@ -585,7 +611,13 @@ def assemble_rung_evidence( def ingest_benchexec_result( - *, root: Path, stage: Path, scale: int, plan: Mapping[str, Any] + *, + root: Path, + stage: Path, + scale: int, + plan: Mapping[str, Any], + profile_id: str | None = None, + source: str = "progressive_profile", ) -> tuple[dict[str, Any], Mapping[str, Any], dict[str, Any]]: raw_output = stage / "raw" graphforge = _parse_graphforge_log(raw_output) @@ -604,7 +636,12 @@ def ingest_benchexec_result( _validate(root, "certification-evidence.json", graphforge) _validate(root, "benchexec-run-evidence.json", benchexec) rung = assemble_rung_evidence( - root=root, scale=scale, graphforge=graphforge, benchexec=benchexec + root=root, + scale=scale, + graphforge=graphforge, + benchexec=benchexec, + profile_id=profile_id, + source=source, ) return benchexec, graphforge, rung diff --git a/benchmarks/harness/graphforge_bench/qualification_operator.py b/benchmarks/harness/graphforge_bench/qualification_operator.py index 2463f704..7a466720 100644 --- a/benchmarks/harness/graphforge_bench/qualification_operator.py +++ b/benchmarks/harness/graphforge_bench/qualification_operator.py @@ -3,8 +3,8 @@ The canonical ``run`` command opens an ESC environment and invokes the existing controller inside Pulumi's secret-filtered process. It does not claim that a caller-controlled marker can prove ESC ancestry. The progressive ladder remains -fail-closed until a dedicated scale executor exists; its no-spend plan is -available through ``plan-progressive``. +fail-closed until whole-attempt orchestration, spend authorization, and teardown +recovery exist; its no-spend plan is available through ``plan-progressive``. """ from __future__ import annotations @@ -52,8 +52,8 @@ def _single_value(argv: Sequence[str], flag: str) -> str: def validate_live_request(gate: str, argv: Sequence[str]) -> None: if gate == "progressive-ladder": raise OperatorRefusalError( - "progressive-ladder execution is unavailable until the dedicated provider " - "image and BenchExec scale executor are implemented" + "progressive-ladder execution is unavailable until whole-attempt Fly orchestration, " + "spend authorization, and teardown recovery are implemented" ) if gate not in LIVE_GATES: raise OperatorRefusalError("qualification gate is unknown") diff --git a/benchmarks/schemas/progressive-provider-plan.json b/benchmarks/schemas/progressive-provider-plan.json index 6846b60b..c6fb676b 100644 --- a/benchmarks/schemas/progressive-provider-plan.json +++ b/benchmarks/schemas/progressive-provider-plan.json @@ -3,7 +3,7 @@ "schema": "graphforge-progressive-provider-plan-schema/1", "type": "object", "additionalProperties": false, - "required": ["schema", "status", "commit", "maximum_scale", "completed_scales", "next_rung", "execution", "profile_id", "profile_path", "profile_sha256", "projection", "execution_authorized", "execution_refusal", "claim"], + "required": ["schema", "status", "commit", "maximum_scale", "completed_scales", "next_rung", "execution", "profile_id", "profile_path", "profile_sha256", "image_digest", "projection", "execution_authorized", "execution_refusal", "claim"], "properties": { "schema": {"const": "graphforge-progressive-provider-plan/1"}, "status": {"const": "admitted"}, @@ -15,6 +15,12 @@ "profile_id": {"type": "string", "pattern": "^graph500-s(18|19|20|22|24|25|26)-(local|provider)$"}, "profile_path": {"type": "string", "pattern": "^profiles/graph500/s(18|19|20|22|24|25|26)-(local|provider)\\.json$"}, "profile_sha256": {"type": "string", "pattern": "^sha256:[0-9a-f]{64}$"}, + "image_digest": { + "oneOf": [ + {"type": "null"}, + {"type": "string", "pattern": "^registry\\.fly\\.io/[a-z0-9][a-z0-9._/-]*@sha256:[0-9a-f]{64}$"} + ] + }, "projection": {"oneOf": [{"type": "object"}, {"type": "null"}]}, "execution_authorized": {"type": "boolean"}, "execution_refusal": {"type": ["string", "null"], "enum": [null, "provider_executor_unavailable"]}, diff --git a/benchmarks/schemas/progressive-provider-run-plan.json b/benchmarks/schemas/progressive-provider-run-plan.json new file mode 100644 index 00000000..6598034b --- /dev/null +++ b/benchmarks/schemas/progressive-provider-run-plan.json @@ -0,0 +1,50 @@ +{ + "$schema": "https://json-schema.org/draft/2020-12/schema", + "schema": "graphforge-progressive-provider-run-plan-schema/1", + "type": "object", + "additionalProperties": false, + "required": ["schema", "rung", "execution", "identities", "limits", "outputs", "claim"], + "properties": { + "schema": {"const": "graphforge-progressive-provider-execution-plan/1"}, + "rung": {"enum": ["S20", "S22", "S24", "S25", "S26"]}, + "execution": {"const": "provider_native_linux_benchexec"}, + "identities": { + "type": "object", + "additionalProperties": false, + "required": ["commit", "profile_id", "profile_sha256", "image_digest", "generator", "generator_executable_sha256", "gf_sha256", "certify_sha256", "benchexec_python_sha256", "benchexec_version", "admitted_plan_sha256", "source_tree_sha256"], + "properties": { + "commit": {"type": "string", "pattern": "^[0-9a-f]{40}$"}, + "profile_id": {"enum": ["graph500-s20-provider", "graph500-s22-provider", "graph500-s24-provider", "graph500-s25-provider", "graph500-s26-provider"]}, + "profile_sha256": {"$ref": "#/$defs/digest"}, + "image_digest": {"type": "string", "pattern": "^registry\\.fly\\.io/[a-z0-9][a-z0-9._/-]*@sha256:[0-9a-f]{64}$"}, + "generator": {"type": "string", "pattern": "^sha256:[0-9a-f]{64}$"}, + "generator_executable_sha256": {"$ref": "#/$defs/digest"}, + "gf_sha256": {"$ref": "#/$defs/digest"}, + "certify_sha256": {"$ref": "#/$defs/digest"}, + "benchexec_python_sha256": {"$ref": "#/$defs/digest"}, + "benchexec_version": {"type": "string", "pattern": "^[0-9]+(\\.[0-9]+)+$"}, + "admitted_plan_sha256": {"$ref": "#/$defs/digest"}, + "source_tree_sha256": {"$ref": "#/$defs/digest"} + } + }, + "limits": { + "type": "object", + "additionalProperties": false, + "required": ["wall_seconds", "memory_bytes", "cores"], + "properties": { + "wall_seconds": {"const": 14400}, + "memory_bytes": {"const": 4294967296}, + "cores": {"const": 16} + } + }, + "outputs": { + "type": "array", + "minItems": 5, + "maxItems": 5, + "uniqueItems": true, + "items": {"type": "string", "pattern": "^s(20|22|24|25|26)-(plan|benchexec|graphforge|rung|result)\\.json$"} + }, + "claim": {"const": "engineering_evidence_only"} + }, + "$defs": {"digest": {"type": "string", "pattern": "^[0-9a-f]{64}$"}} +} diff --git a/benchmarks/schemas/progressive-provider-run-result.json b/benchmarks/schemas/progressive-provider-run-result.json index 6cd0b52e..4eb34b3e 100644 --- a/benchmarks/schemas/progressive-provider-run-result.json +++ b/benchmarks/schemas/progressive-provider-run-result.json @@ -3,27 +3,55 @@ "schema": "graphforge-progressive-provider-run-result-schema/1", "type": "object", "additionalProperties": false, - "required": ["schema", "rung", "status", "failure", "identities", "claim"], + "required": ["schema", "rung", "status", "failure", "identities", "artifacts", "claim"], "properties": { "schema": {"const": "graphforge-progressive-provider-run-result/1"}, "rung": {"enum": ["S20", "S22", "S24", "S25", "S26"]}, "status": {"enum": ["passed", "failed"]}, - "failure": {"type": ["string", "null"]}, + "failure": { + "enum": [null, "native_authority_unavailable", "staging_failed", "execution_boundary_failed", "benchexec_failed", "ordinary_receipt_missing"] + }, "identities": { "type": "object", "additionalProperties": false, - "required": ["commit", "profile_id", "profile_sha256"], + "required": ["commit", "profile_id", "profile_sha256", "image_digest", "generator", "generator_executable_sha256", "gf_sha256", "certify_sha256", "benchexec_python_sha256", "benchexec_version", "admitted_plan_sha256", "source_tree_sha256"], "properties": { "commit": {"type": "string", "pattern": "^[0-9a-f]{40}$"}, "profile_id": {"enum": ["graph500-s20-provider", "graph500-s22-provider", "graph500-s24-provider", "graph500-s25-provider", "graph500-s26-provider"]}, - "profile_sha256": {"type": "string", "pattern": "^[0-9a-f]{64}$"} + "profile_sha256": {"$ref": "#/$defs/digest"}, + "image_digest": {"type": "string", "pattern": "^registry\\.fly\\.io/[a-z0-9][a-z0-9._/-]*@sha256:[0-9a-f]{64}$"}, + "generator": {"type": "string", "pattern": "^sha256:[0-9a-f]{64}$"}, + "generator_executable_sha256": {"$ref": "#/$defs/digest"}, + "gf_sha256": {"$ref": "#/$defs/digest"}, + "certify_sha256": {"$ref": "#/$defs/digest"}, + "benchexec_python_sha256": {"$ref": "#/$defs/digest"}, + "benchexec_version": {"type": "string", "pattern": "^[0-9]+(\\.[0-9]+)+$"}, + "admitted_plan_sha256": {"$ref": "#/$defs/digest"}, + "source_tree_sha256": {"$ref": "#/$defs/digest"} } }, + "artifacts": { + "oneOf": [ + {"type": "null"}, + { + "type": "object", + "additionalProperties": false, + "required": ["plan_sha256", "benchexec_sha256", "graphforge_sha256", "rung_sha256"], + "properties": { + "plan_sha256": {"$ref": "#/$defs/digest"}, + "benchexec_sha256": {"$ref": "#/$defs/digest"}, + "graphforge_sha256": {"$ref": "#/$defs/digest"}, + "rung_sha256": {"$ref": "#/$defs/digest"} + } + } + ] + }, "claim": {"const": "engineering_evidence_only"} }, "allOf": [ {"if": {"properties": {"status": {"const": "passed"}}, "required": ["status"]}, - "then": {"properties": {"failure": {"type": "null"}}}, - "else": {"properties": {"failure": {"type": "string"}}}} - ] + "then": {"properties": {"failure": {"type": "null"}, "artifacts": {"type": "object"}}}, + "else": {"properties": {"failure": {"type": "string"}, "artifacts": {"type": "null"}}}} + ], + "$defs": {"digest": {"type": "string", "pattern": "^[0-9a-f]{64}$"}} } diff --git a/benchmarks/tests/test_progressive_provider_plan.py b/benchmarks/tests/test_progressive_provider_plan.py index 15b7548e..81f31bbf 100644 --- a/benchmarks/tests/test_progressive_provider_plan.py +++ b/benchmarks/tests/test_progressive_provider_plan.py @@ -155,6 +155,7 @@ def result(scale: int) -> dict: "io_reader_publication_headroom", "correctness", ) +IMAGE = "registry.fly.io/graphforge-bench@sha256:" + "1" * 64 class ProgressiveProviderPlanTests(unittest.TestCase): @@ -172,6 +173,7 @@ def test_empty_workspace_admits_only_s18(self) -> None: self.assertEqual(plan["next_rung"], "S18") self.assertEqual(plan["execution"], "local") self.assertIsNone(plan["projection"]) + self.assertIsNone(plan["image_digest"]) self.assertEqual(plan["profile_path"], "profiles/graph500/s18-local.json") self.assertNotIn("app", json.dumps(plan)) self.assertNotIn("machine", json.dumps(plan)) @@ -197,6 +199,14 @@ def test_evidence_must_be_a_contiguous_passed_prefix(self) -> None: with self.assertRaisesRegex(ProviderPlanError, "schema-valid"): completed_rungs(ROOT, self.output) + (self.output / "s18-rung.json").unlink() + for scale in (18, 19, 20): + (self.output / f"s{scale}-rung.json").write_text( + json.dumps(rung(scale, source="progressive_profile")), encoding="utf-8" + ) + with self.assertRaisesRegex(ProviderPlanError, "canonical profile"): + completed_rungs(ROOT, self.output) + def test_projection_gate_is_required_before_s20(self) -> None: with ( patch( @@ -246,9 +256,11 @@ def test_provider_plan_contains_only_sanitized_projection(self) -> None: commit=COMMIT, maximum_scale=20, provider_capacity=CAPACITY, + image_digest=IMAGE, ) self.assertEqual(plan["next_rung"], "S20") self.assertEqual(plan["profile_id"], "graph500-s20-provider") + self.assertEqual(plan["image_digest"], IMAGE) self.assertEqual(plan["projection"]["decision"], "admitted") encoded = json.dumps(plan).lower() for forbidden in ("machine_id", "volume_id", "token", "secret", "provider_id"): @@ -313,7 +325,87 @@ def test_minimal_result_document_is_rejected(self) -> None: provider_capacity=CAPACITY, ) - def test_provider_execution_is_explicitly_refused(self) -> None: + def test_provider_result_hashes_bind_every_projection_artifact(self) -> None: + for scale in (18, 19): + (self.output / f"s{scale}-rung.json").write_text( + json.dumps(rung(scale)), encoding="utf-8" + ) + (self.output / f"s{scale}-result.json").write_text( + json.dumps(result(scale)), encoding="utf-8" + ) + scale = 20 + artifact_paths = { + "plan_sha256": self.output / "s20-plan.json", + "benchexec_sha256": self.output / "s20-benchexec.json", + "graphforge_sha256": self.output / "s20-graphforge.json", + "rung_sha256": self.output / "s20-rung.json", + } + profile = ROOT / "profiles/graph500/s20-provider.json" + identities = { + "commit": COMMIT, + "profile_id": "graph500-s20-provider", + "profile_sha256": hashlib.sha256(profile.read_bytes()).hexdigest(), + "image_digest": IMAGE, + "generator": "sha256:" + "0" * 64, + "generator_executable_sha256": "0" * 64, + "gf_sha256": "0" * 64, + "certify_sha256": "0" * 64, + "benchexec_python_sha256": "0" * 64, + "benchexec_version": "1.0", + "admitted_plan_sha256": "0" * 64, + "source_tree_sha256": "0" * 64, + } + execution_plan = { + "schema": "graphforge-progressive-provider-execution-plan/1", + "rung": "S20", + "execution": "provider_native_linux_benchexec", + "identities": identities, + "limits": {"wall_seconds": 14400, "memory_bytes": 4294967296, "cores": 16}, + "outputs": [ + "s20-plan.json", + "s20-benchexec.json", + "s20-graphforge.json", + "s20-rung.json", + "s20-result.json", + ], + "claim": "engineering_evidence_only", + } + for name, path in artifact_paths.items(): + if name == "plan_sha256": + document = execution_plan + elif name == "rung_sha256": + document = rung(scale) + else: + document = {"fixture": name} + path.write_text(json.dumps(document), encoding="utf-8") + provider_result = { + "schema": "graphforge-progressive-provider-run-result/1", + "rung": "S20", + "status": "passed", + "failure": None, + "identities": identities, + "artifacts": { + name: hashlib.sha256(path.read_bytes()).hexdigest() + for name, path in artifact_paths.items() + }, + "claim": "engineering_evidence_only", + } + (self.output / "s20-result.json").write_text(json.dumps(provider_result), encoding="utf-8") + self.assertEqual(len(completed_rungs(ROOT, self.output, commit=COMMIT)), 3) + + changed_result = json.loads(json.dumps(provider_result)) + changed_result["identities"]["gf_sha256"] = "1" * 64 + (self.output / "s20-result.json").write_text(json.dumps(changed_result), encoding="utf-8") + with self.assertRaisesRegex(ProviderPlanError, "plan/result identities disagree"): + completed_rungs(ROOT, self.output, commit=COMMIT) + (self.output / "s20-result.json").write_text(json.dumps(provider_result), encoding="utf-8") + + with (self.output / "s20-benchexec.json").open("a", encoding="utf-8") as stream: + stream.write("\n") + with self.assertRaisesRegex(ProviderPlanError, "artifacts do not match"): + completed_rungs(ROOT, self.output, commit=COMMIT) + + def test_provider_plan_has_offline_execution_authority(self) -> None: with patch( "graphforge_bench.progressive_provider_plan.completed_rungs", return_value=[rung(18), rung(19)], @@ -324,9 +416,28 @@ def test_provider_execution_is_explicitly_refused(self) -> None: commit=COMMIT, maximum_scale=20, provider_capacity=CAPACITY, + image_digest=IMAGE, + ) + require_execution_authority(plan) + refused = {**plan, "execution_authorized": False, "execution_refusal": "unavailable"} + with self.assertRaisesRegex(ProviderPlanError, "authority is unavailable"): + require_execution_authority(refused) + + def test_provider_plan_requires_one_immutable_admitted_image(self) -> None: + with ( + patch( + "graphforge_bench.progressive_provider_plan.completed_rungs", + return_value=[rung(18), rung(19)], + ), + self.assertRaisesRegex(ProviderPlanError, "immutable provider image digest"), + ): + plan_provider_ladder( + root=ROOT, + output_dir=self.output, + commit=COMMIT, + maximum_scale=20, + provider_capacity=CAPACITY, ) - with self.assertRaisesRegex(ProviderPlanError, "dedicated provider image"): - require_execution_authority(plan) if __name__ == "__main__": diff --git a/benchmarks/tests/test_progressive_provider_run.py b/benchmarks/tests/test_progressive_provider_run.py new file mode 100644 index 00000000..8c8de3f0 --- /dev/null +++ b/benchmarks/tests/test_progressive_provider_run.py @@ -0,0 +1,584 @@ +from __future__ import annotations + +import hashlib +import json +from pathlib import Path +import subprocess +import tempfile +import unittest +from unittest.mock import Mock, patch + +from graphforge_bench.progressive_provider_run import ( + ProviderRunError, + _execution_commit, + _schema, + build_execution_plan, + main, + run, + validate_admitted_plan, +) +from graphforge_bench.progressive_run import ControllerError, Executables +from tests.test_progressive_run import ( + benchexec as benchexec_fixture, +) +from tests.test_progressive_run import ( + graphforge as graphforge_fixture, +) +from tests.test_progressive_run import ( + passed_rung as rung_fixture, +) + +ROOT = Path(__file__).resolve().parents[1] +COMMIT = subprocess.run( + ["git", "-C", str(ROOT.parent), "rev-parse", "HEAD"], + capture_output=True, + check=True, + text=True, +).stdout.strip() +IMAGE = "registry.fly.io/graphforge-bench@sha256:" + "1" * 64 +PROJECTED_FIELDS = ( + "wall_seconds", + "peak_rss_bytes", + "retained_storage_bytes", + "transient_peak_storage_bytes", + "logical_read_bytes", + "logical_write_bytes", + "physical_read_bytes", + "physical_write_bytes", + "reader_calls", + "publication_work_units", + "storage_peak_bytes", +) +RATE_FIELDS = ( + "physical_read_bytes_per_second", + "physical_write_bytes_per_second", + "reader_calls_per_second", + "publication_work_per_second", +) +SLOPE_FIELDS = ( + "logical_read_bytes", + "logical_write_bytes", + "physical_read_bytes", + "physical_write_bytes", + "reader_calls", + "publication_work_units", +) +CHECK_FIELDS = ( + "time_headroom", + "rss_headroom", + "retained_storage_headroom", + "transient_storage_headroom", + "storage_headroom", + "rss_bounded_or_plateaued", + "io_reader_publication_capacity_measured", + "io_reader_publication_headroom", + "correctness", +) + + +def admitted_plan(scale: int = 20) -> dict: + profile = ROOT / "profiles" / "graph500" / f"s{scale}-provider.json" + ladder = [18, 19, 20, 22, 24, 25, 26] + sources = {20: [18, 19], 22: [19, 20], 24: [20, 22], 25: [22, 24], 26: [24, 25]} + return { + "schema": "graphforge-progressive-provider-plan/1", + "status": "admitted", + "commit": COMMIT, + "maximum_scale": 26, + "completed_scales": ladder[: ladder.index(scale)], + "next_rung": f"S{scale}", + "execution": "provider", + "profile_id": f"graph500-s{scale}-provider", + "profile_path": f"profiles/graph500/s{scale}-provider.json", + "profile_sha256": "sha256:" + hashlib.sha256(profile.read_bytes()).hexdigest(), + "image_digest": IMAGE, + "projection": { + "schema": "graphforge-progressive-qualification-evidence/1", + "target": f"S{scale}", + "source_scales": sources[scale], + "decision": "admitted", + "limits": { + "wall_seconds": 14_400, + "rss_bytes": 4_294_967_296, + "volume_bytes": 536_870_912_000, + }, + "headroom": { + "time_fraction": 0.2, + "rss_fraction": 0.2, + "storage_fraction": 0.15, + }, + "projected": dict.fromkeys(PROJECTED_FIELDS, 1), + "required_rates": dict.fromkeys(RATE_FIELDS, 1), + "provider_capacity": dict.fromkeys(RATE_FIELDS, 100), + "slopes_observed": dict.fromkeys(SLOPE_FIELDS, 1), + "rss_growth_fraction": 0, + "checks": dict.fromkeys(CHECK_FIELDS, True), + "claim": "engineering_evidence_only", + }, + "execution_authorized": True, + "execution_refusal": None, + "claim": "engineering_evidence_only", + } + + +class ProgressiveProviderRunTests(unittest.TestCase): + def setUp(self) -> None: + self.temporary = tempfile.TemporaryDirectory() + self.base = Path(self.temporary.name) + self.output = self.base / "evidence" + generator = self.base / "graphforge-benchmark-graph500-generator" + generator.write_bytes((ROOT / "runners/graph500-generator/src/main.rs").read_bytes()) + gf = self.base / "gf" + certify = self.base / "graphforge-benchmark-certify" + python = self.base / "python" + for path in (gf, certify, python): + path.write_bytes(b"immutable fixture executable") + for path in (generator, gf, certify, python): + path.chmod(0o755) + self.executables = Executables(gf, certify, generator, python) + + def tearDown(self) -> None: + self.temporary.cleanup() + + def execution_plan(self) -> dict: + with patch( + "graphforge_bench.progressive_provider_run._benchexec_version", + return_value="3.30", + ): + return build_execution_plan( + root=ROOT, + admitted_plan=admitted_plan(), + admitted_plan_sha256="a" * 64, + image_digest=IMAGE, + executables=self.executables, + source_tree_sha256="b" * 64, + ) + + def test_plan_binds_admission_image_and_all_native_identities(self) -> None: + plan = self.execution_plan() + identities = plan["identities"] + self.assertEqual(plan["rung"], "S20") + self.assertEqual(identities["commit"], COMMIT) + self.assertEqual(identities["image_digest"], IMAGE) + self.assertEqual(identities["admitted_plan_sha256"], "a" * 64) + self.assertEqual( + identities["profile_sha256"], + hashlib.sha256((ROOT / "profiles/graph500/s20-provider.json").read_bytes()).hexdigest(), + ) + _schema(ROOT, "progressive-provider-run-plan.json", plan) + + def test_image_commit_attestation_works_without_git_metadata(self) -> None: + image_root = self.base / "image" / "benchmarks" + image_root.mkdir(parents=True) + (image_root.parent / "commit").write_text(COMMIT + "\n", encoding="ascii") + with patch( + "graphforge_bench.progressive_provider_run.repository_commit", + side_effect=AssertionError("image execution must not require .git"), + ): + self.assertEqual(_execution_commit(image_root), COMMIT) + + (image_root.parent / "commit").write_text("not-a-commit\n", encoding="ascii") + with self.assertRaisesRegex(ProviderRunError, "malformed"): + _execution_commit(image_root) + + def test_production_cli_has_no_laptop_or_executable_override_fallback(self) -> None: + with ( + patch( + "graphforge_bench.progressive_provider_run._require_work_mount", + side_effect=ProviderRunError("provider work volume is unavailable"), + ), + patch( + "graphforge_bench.progressive_provider_run._read_document", + side_effect=AssertionError("must refuse before reading a caller plan"), + ), + patch( + "graphforge_bench.progressive_provider_run.resolve_executables", + side_effect=AssertionError("must refuse before selecting executables"), + ), + ): + self.assertEqual( + main( + [ + "--admitted-plan", + str(self.base / "plan.json"), + "--output-dir", + str(self.output), + "--image-digest", + IMAGE, + ] + ), + 2, + ) + with self.assertRaises(SystemExit): + main( + [ + "--admitted-plan", + str(self.base / "plan.json"), + "--output-dir", + str(self.output), + "--image-digest", + IMAGE, + "--gf", + str(self.executables.gf), + ] + ) + + def test_each_provider_rung_has_one_closed_offline_plan(self) -> None: + for scale in (20, 22, 24, 25, 26): + with ( + self.subTest(scale=scale), + patch( + "graphforge_bench.progressive_provider_run._benchexec_version", + return_value="3.30", + ), + ): + plan = build_execution_plan( + root=ROOT, + admitted_plan=admitted_plan(scale), + admitted_plan_sha256="a" * 64, + image_digest=IMAGE, + executables=self.executables, + source_tree_sha256="b" * 64, + ) + self.assertEqual(plan["rung"], f"S{scale}") + self.assertEqual(plan["identities"]["profile_id"], f"graph500-s{scale}-provider") + + def test_tampered_plan_and_wrong_rung_are_refused(self) -> None: + tampered = admitted_plan() + tampered["profile_sha256"] = "sha256:" + "0" * 64 + with self.assertRaisesRegex(ProviderRunError, "identity or projection"): + validate_admitted_plan(ROOT, tampered) + wrong = admitted_plan() + wrong["next_rung"] = "S18" + wrong["profile_id"] = "graph500-s18-local" + wrong["profile_path"] = "profiles/graph500/s18-local.json" + with self.assertRaisesRegex(ProviderRunError, "provider rung|validation failed"): + validate_admitted_plan(ROOT, wrong) + + for scale in (22, 24, 25, 26): + expected = admitted_plan(scale) + for completed in ( + expected["completed_scales"][:-1], + list(reversed(expected["completed_scales"])), + [*expected["completed_scales"], scale], + ): + with self.subTest(scale=scale, completed=completed): + mutated = {**expected, "completed_scales": completed} + with self.assertRaisesRegex(ProviderRunError, "identity or projection"): + validate_admitted_plan(ROOT, mutated) + + def test_admitted_image_must_match_the_running_image(self) -> None: + with ( + patch( + "graphforge_bench.progressive_provider_run._benchexec_version", + return_value="3.30", + ), + self.assertRaisesRegex(ProviderRunError, "does not match"), + ): + build_execution_plan( + root=ROOT, + admitted_plan=admitted_plan(), + admitted_plan_sha256="a" * 64, + image_digest="registry.fly.io/graphforge-bench@sha256:" + "2" * 64, + executables=self.executables, + source_tree_sha256="b" * 64, + ) + + def test_execution_requires_explicit_plan_authority(self) -> None: + refused = admitted_plan() + refused["execution_authorized"] = False + refused["execution_refusal"] = "provider_executor_unavailable" + with self.assertRaisesRegex(ProviderRunError, "identity or projection"): + validate_admitted_plan(ROOT, refused) + + def test_identity_drift_refuses_before_authority_or_execution(self) -> None: + plan = self.execution_plan() + self.executables.gf.write_bytes(b"changed after planning") + execute = Mock(return_value=0) + authority = Mock(return_value={"result": "passed"}) + with ( + patch( + "graphforge_bench.progressive_provider_run._benchexec_version", + return_value="3.30", + ), + self.assertRaisesRegex(ProviderRunError, "changed after planning"), + ): + run( + root=ROOT, + output_dir=self.output, + plan=plan, + executables=self.executables, + expected_image_digest=IMAGE, + expected_admitted_plan_sha256="a" * 64, + expected_source_tree_sha256="b" * 64, + execution_boundary=execute, + authority_boundary=authority, + ) + authority.assert_not_called() + execute.assert_not_called() + self.assertFalse(self.output.exists()) + + def test_admission_identity_drift_refuses_before_authority(self) -> None: + for identity, value in ( + ("image_digest", "registry.fly.io/graphforge-bench@sha256:" + "2" * 64), + ("admitted_plan_sha256", "c" * 64), + ("source_tree_sha256", "d" * 64), + ): + with self.subTest(identity=identity): + plan = self.execution_plan() + plan["identities"][identity] = value + authority = Mock(return_value={"result": "passed"}) + with ( + patch( + "graphforge_bench.progressive_provider_run._benchexec_version", + return_value="3.30", + ), + self.assertRaisesRegex(ProviderRunError, "changed after planning"), + ): + run( + root=ROOT, + output_dir=self.output, + plan=plan, + executables=self.executables, + expected_image_digest=IMAGE, + expected_admitted_plan_sha256="a" * 64, + expected_source_tree_sha256="b" * 64, + execution_boundary=Mock(return_value=0), + authority_boundary=authority, + ) + authority.assert_not_called() + + def test_boundary_exceptions_emit_closed_failures(self) -> None: + cases = ( + ("authority", lambda: (_ for _ in ()).throw(OSError("fixture"))), + ("execution", lambda: {"result": "passed"}), + ) + for boundary, authority in cases: + with self.subTest(boundary=boundary): + plan = self.execution_plan() + execution = ( + (lambda *_: (_ for _ in ()).throw(OSError("fixture"))) + if boundary == "execution" + else Mock(return_value=0) + ) + with ( + patch( + "graphforge_bench.progressive_provider_run._benchexec_version", + return_value="3.30", + ), + self.assertRaises(ProviderRunError), + ): + run( + root=ROOT, + output_dir=self.output, + plan=plan, + executables=self.executables, + expected_image_digest=IMAGE, + expected_admitted_plan_sha256="a" * 64, + expected_source_tree_sha256="b" * 64, + execution_boundary=execution, + authority_boundary=authority, + ) + result = json.loads((self.output / "s20-result.json").read_text()) + self.assertIn( + result["failure"], + {"native_authority_unavailable", "execution_boundary_failed"}, + ) + _schema(ROOT, "progressive-provider-run-result.json", result) + for path in self.output.iterdir(): + if path.is_file(): + path.unlink() + + def test_no_native_authority_refuses_without_execution(self) -> None: + plan = self.execution_plan() + execute = Mock(return_value=0) + with ( + patch( + "graphforge_bench.progressive_provider_run._benchexec_version", + return_value="3.30", + ), + self.assertRaisesRegex(ProviderRunError, "authority is unavailable"), + ): + run( + root=ROOT, + output_dir=self.output, + plan=plan, + executables=self.executables, + expected_image_digest=IMAGE, + expected_admitted_plan_sha256="a" * 64, + expected_source_tree_sha256="b" * 64, + execution_boundary=execute, + authority_boundary=lambda: {"result": "failed"}, + ) + execute.assert_not_called() + result = json.loads((self.output / "s20-result.json").read_text()) + self.assertEqual(result["failure"], "native_authority_unavailable") + + def test_benchexec_failure_emits_closed_failed_result(self) -> None: + plan = self.execution_plan() + with ( + patch( + "graphforge_bench.progressive_provider_run._benchexec_version", + return_value="3.30", + ), + self.assertRaisesRegex(ProviderRunError, "benchexec_failed"), + ): + run( + root=ROOT, + output_dir=self.output, + plan=plan, + executables=self.executables, + expected_image_digest=IMAGE, + expected_admitted_plan_sha256="a" * 64, + expected_source_tree_sha256="b" * 64, + execution_boundary=lambda *_: 1, + authority_boundary=lambda: {"result": "passed"}, + ) + result = json.loads((self.output / "s20-result.json").read_text()) + self.assertEqual(result["failure"], "benchexec_failed") + _schema(ROOT, "progressive-provider-run-result.json", result) + + def test_missing_receipts_fail_closed(self) -> None: + plan = self.execution_plan() + with ( + patch( + "graphforge_bench.progressive_provider_run._benchexec_version", + return_value="3.30", + ), + patch( + "graphforge_bench.progressive_provider_run.ingest_benchexec_result", + side_effect=ControllerError("missing"), + ), + self.assertRaisesRegex(ProviderRunError, "ordinary_receipt_missing"), + ): + run( + root=ROOT, + output_dir=self.output, + plan=plan, + executables=self.executables, + expected_image_digest=IMAGE, + expected_admitted_plan_sha256="a" * 64, + expected_source_tree_sha256="b" * 64, + execution_boundary=lambda *_: 0, + authority_boundary=lambda: {"result": "passed"}, + ) + result = json.loads((self.output / "s20-result.json").read_text()) + self.assertEqual(result["status"], "failed") + self.assertEqual(result["failure"], "ordinary_receipt_missing") + + def test_success_emits_exactly_five_canonical_engineering_files(self) -> None: + plan = self.execution_plan() + self.output.mkdir() + (self.output / "s20-plan.json").write_text(json.dumps(plan), encoding="utf-8") + graphforge = graphforge_fixture(20) + graphforge["profile_id"] = "graph500-s20-provider" + benchexec = benchexec_fixture(graphforge) + rung = rung_fixture(20) + rung["profile_id"] = "graph500-s20-provider" + rung["source"] = "canonical_ladder" + with ( + patch( + "graphforge_bench.progressive_provider_run._benchexec_version", + return_value="3.30", + ), + patch( + "graphforge_bench.progressive_provider_run.ingest_benchexec_result", + return_value=(benchexec, graphforge, rung), + ) as ingest, + ): + run( + root=ROOT, + output_dir=self.output, + plan=plan, + executables=self.executables, + expected_image_digest=IMAGE, + expected_admitted_plan_sha256="a" * 64, + expected_source_tree_sha256="b" * 64, + execution_boundary=lambda *_: 0, + authority_boundary=lambda: {"result": "passed"}, + ) + self.assertEqual( + {path.name for path in self.output.iterdir()}, + { + "s20-plan.json", + "s20-benchexec.json", + "s20-graphforge.json", + "s20-rung.json", + "s20-result.json", + }, + ) + self.assertEqual( + ingest.call_args.kwargs, + { + "root": ROOT, + "stage": ingest.call_args.kwargs["stage"], + "scale": 20, + "plan": plan, + "profile_id": "graph500-s20-provider", + "source": "canonical_ladder", + }, + ) + result = json.loads((self.output / "s20-result.json").read_text()) + self.assertEqual(result["claim"], "engineering_evidence_only") + + def test_every_provider_rung_writes_schema_valid_hash_bound_result(self) -> None: + for scale in (20, 22, 24, 25, 26): + with self.subTest(scale=scale): + output = self.base / f"s{scale}" + output.mkdir() + with patch( + "graphforge_bench.progressive_provider_run._benchexec_version", + return_value="3.30", + ): + plan = build_execution_plan( + root=ROOT, + admitted_plan=admitted_plan(scale), + admitted_plan_sha256="a" * 64, + image_digest=IMAGE, + executables=self.executables, + source_tree_sha256="b" * 64, + ) + (output / f"s{scale}-plan.json").write_text(json.dumps(plan), encoding="utf-8") + graphforge = graphforge_fixture(scale) + graphforge["profile_id"] = f"graph500-s{scale}-provider" + benchexec = benchexec_fixture(graphforge) + rung = rung_fixture(scale) + rung["profile_id"] = f"graph500-s{scale}-provider" + rung["source"] = "canonical_ladder" + with ( + patch( + "graphforge_bench.progressive_provider_run._benchexec_version", + return_value="3.30", + ), + patch( + "graphforge_bench.progressive_provider_run.ingest_benchexec_result", + return_value=(benchexec, graphforge, rung), + ), + ): + run( + root=ROOT, + output_dir=output, + plan=plan, + executables=self.executables, + expected_image_digest=IMAGE, + expected_admitted_plan_sha256="a" * 64, + expected_source_tree_sha256="b" * 64, + execution_boundary=lambda *_: 0, + authority_boundary=lambda: {"result": "passed"}, + ) + result = json.loads((output / f"s{scale}-result.json").read_text()) + _schema(ROOT, "progressive-provider-run-result.json", result) + for artifact, filename in ( + ("plan_sha256", f"s{scale}-plan.json"), + ("benchexec_sha256", f"s{scale}-benchexec.json"), + ("graphforge_sha256", f"s{scale}-graphforge.json"), + ("rung_sha256", f"s{scale}-rung.json"), + ): + self.assertEqual( + result["artifacts"][artifact], + hashlib.sha256((output / filename).read_bytes()).hexdigest(), + ) + + +if __name__ == "__main__": + unittest.main() diff --git a/benchmarks/tests/test_progressive_run.py b/benchmarks/tests/test_progressive_run.py index e990c220..b6c2d668 100644 --- a/benchmarks/tests/test_progressive_run.py +++ b/benchmarks/tests/test_progressive_run.py @@ -4,6 +4,7 @@ from pathlib import Path import tempfile import unittest +from unittest.mock import patch from graphforge_bench.progressive_run import ( ControllerError, @@ -11,11 +12,13 @@ _run_benchexec, _safe_stage, _validate, + _write_json, assemble_rung_evidence, build_plan, ingest_benchexec_result, require_bulk_ingest_capability, require_order, + resolve_executables, validate_fixture_bundle, write_plan, write_s20_projection, @@ -265,6 +268,28 @@ def setUp(self) -> None: def tearDown(self) -> None: self.temporary.cleanup() + def test_evidence_writes_are_atomic_and_never_replace_existing_files(self) -> None: + path = self.base / "immutable.json" + _write_json(path, {"value": 1}) + with self.assertRaises(FileExistsError): + _write_json(path, {"value": 2}) + self.assertEqual(json.loads(path.read_text()), {"value": 1}) + self.assertEqual([item.name for item in self.base.glob(".immutable.json.*")], []) + + def test_venv_python_path_is_not_dereferenced_out_of_its_environment(self) -> None: + venv = self.base / "venv/bin" + venv.mkdir(parents=True) + python = venv / "python" + python.symlink_to(self.executables.benchexec_python) + resolved = resolve_executables( + gf=str(self.executables.gf), + certify=str(self.executables.certify), + generator=str(self.executables.generator), + benchexec_python=str(python), + ) + self.assertEqual(resolved.benchexec_python, python.absolute()) + self.assertTrue(resolved.benchexec_python.is_symlink()) + def test_dry_plan_binds_exact_immutable_identities_without_paths(self) -> None: plan = build_plan( root=ROOT, @@ -509,6 +534,25 @@ def test_benchexec_python_is_rechecked_immediately_before_invocation(self) -> No with self.assertRaisesRegex(ControllerError, "identity changed after planning"): _run_benchexec(stage, self.executables, plan["identities"]) + def test_benchexec_sanitized_environment_keeps_only_the_tool_module_path(self) -> None: + plan = build_plan( + root=ROOT, + output_dir=self.output, + scale=18, + commit=COMMIT, + executables=self.executables, + ) + stage = self.base / "benchexec-stage" + stage.mkdir() + (stage / "bin").mkdir() + (stage / "benchmark.xml").write_text("fixture", encoding="utf-8") + with patch("graphforge_bench.progressive_run.subprocess.run") as execute: + execute.return_value.returncode = 0 + self.assertEqual(_run_benchexec(stage, self.executables, plan["identities"]), 0) + environment = execute.call_args.kwargs["env"] + self.assertEqual(environment["PYTHONPATH"], str(ROOT / "harness")) + self.assertEqual(set(environment), {"HOME", "LANG", "LC_ALL", "PATH", "PYTHONPATH"}) + def test_failed_result_schema_requires_closed_exact_identities(self) -> None: plan = build_plan( root=ROOT, diff --git a/benchmarks/tests/test_qualification_operator.py b/benchmarks/tests/test_qualification_operator.py index ac31c61b..4ab77052 100644 --- a/benchmarks/tests/test_qualification_operator.py +++ b/benchmarks/tests/test_qualification_operator.py @@ -61,7 +61,7 @@ def runner(*args: object, **_kwargs: object) -> subprocess.CompletedProcess[str] runner_called = True return subprocess.CompletedProcess(args, 0) - with self.assertRaisesRegex(OperatorRefusalError, "scale executor"): + with self.assertRaisesRegex(OperatorRefusalError, "whole-attempt Fly orchestration"): run_under_esc( "curatelabs/graphforge/qualification", "progressive-ladder", diff --git a/containers/graphforge-progressive-qualification/Dockerfile b/containers/graphforge-progressive-qualification/Dockerfile new file mode 100644 index 00000000..08c66524 --- /dev/null +++ b/containers/graphforge-progressive-qualification/Dockerfile @@ -0,0 +1,73 @@ +FROM rust:1.96-bookworm@sha256:a339861ae23e9abb272cea45dfafde21760d2ce6577a70f8a926153677902663 AS rust-build + +ARG GRAPHFORGE_COMMIT +ARG TARGETARCH +ARG TARGETOS +WORKDIR /source +COPY . . +RUN test "${TARGETOS}/${TARGETARCH}" = linux/amd64 \ + && printf '%s\n' "${GRAPHFORGE_COMMIT}" | grep -Eq '^[0-9a-f]{40}$' \ + && printf '%s\n' "${GRAPHFORGE_COMMIT}" > /graphforge-commit \ + && chmod 0444 /graphforge-commit \ + && tar --sort=name --mtime='UTC 1970-01-01' --owner=0 --group=0 --numeric-owner \ + -cf /tmp/source-tree.tar . \ + && sha256sum /tmp/source-tree.tar | cut -d ' ' -f 1 > /source-tree-sha256 \ + && rm /tmp/source-tree.tar \ + && chmod 0444 /source-tree-sha256 \ + && cargo build --locked --release --package graphforge-cli --bin gf \ + && cargo build --manifest-path benchmarks/Cargo.toml --locked --release \ + --package graphforge-benchmark-certify \ + --package graphforge-benchmark-graph500-generator + + +FROM python:3.12.11-slim-bookworm@sha256:519591d6871b7bc437060736b9f7456b8731f1499a57e22e6c285135ae657bf7 AS python-deps + +ENV UV_LINK_MODE=copy +WORKDIR /opt/graphforge/benchmarks +RUN python -m pip install --no-cache-dir "uv==0.11.33" +COPY benchmarks/pyproject.toml benchmarks/uv.lock ./ +RUN uv sync --frozen --no-dev --no-install-project + + +FROM python:3.12.11-slim-bookworm@sha256:519591d6871b7bc437060736b9f7456b8731f1499a57e22e6c285135ae657bf7 AS qualification + +ARG GRAPHFORGE_COMMIT +LABEL org.opencontainers.image.revision="${GRAPHFORGE_COMMIT}" +RUN mkdir -p /opt/graphforge /work \ + && chown 10001:10001 /work \ + && chmod 0755 /work + +COPY --from=rust-build /source/target/release/gf /usr/local/bin/gf +COPY --from=rust-build /source/benchmarks/target/release/graphforge-benchmark-certify \ + /usr/local/bin/graphforge-benchmark-certify +COPY --from=rust-build /source/benchmarks/target/release/graphforge-benchmark-graph500-generator \ + /usr/local/bin/graphforge-benchmark-graph500-generator +COPY --from=rust-build --chmod=0444 /graphforge-commit /opt/graphforge/commit +COPY --from=python-deps /opt/graphforge/benchmarks/.venv /opt/graphforge/benchmarks/.venv +COPY benchmarks/harness /opt/graphforge/benchmarks/harness +COPY benchmarks/profiles /opt/graphforge/benchmarks/profiles +COPY benchmarks/schemas /opt/graphforge/benchmarks/schemas +COPY benchmarks/definitions /opt/graphforge/benchmarks/definitions +COPY --chmod=0555 containers/graphforge-progressive-qualification/run-qualification.py \ + /usr/local/bin/run-progressive-qualification +COPY --from=rust-build /source-tree-sha256 /tmp/source-tree-sha256 + +RUN gf_sha256="$(sha256sum /usr/local/bin/gf | cut -d ' ' -f 1)" \ + && certify_sha256="$(sha256sum /usr/local/bin/graphforge-benchmark-certify | cut -d ' ' -f 1)" \ + && generator_sha256="$(sha256sum /usr/local/bin/graphforge-benchmark-graph500-generator | cut -d ' ' -f 1)" \ + && benchexec_python_sha256="$(sha256sum /opt/graphforge/benchmarks/.venv/bin/python | cut -d ' ' -f 1)" \ + && source_tree_sha256="$(cat /tmp/source-tree-sha256)" \ + && printf '%s\n' \ + "{\"schema\":\"graphforge-progressive-provider-build/1\",\"commit\":\"${GRAPHFORGE_COMMIT}\",\"source_tree_sha256\":\"${source_tree_sha256}\",\"executables\":{\"gf_sha256\":\"${gf_sha256}\",\"certify_sha256\":\"${certify_sha256}\",\"generator_executable_sha256\":\"${generator_sha256}\",\"benchexec_python_sha256\":\"${benchexec_python_sha256}\"}}" \ + > /opt/graphforge/build-manifest.json \ + && chmod 0444 /opt/graphforge/build-manifest.json \ + && rm /tmp/source-tree-sha256 + +ENV PATH="/opt/graphforge/benchmarks/.venv/bin:${PATH}" \ + PYTHONPATH="/opt/graphforge/benchmarks/harness" \ + PYTHONDONTWRITEBYTECODE=1 \ + PYTHONUNBUFFERED=1 + +WORKDIR / +ENTRYPOINT ["/usr/local/bin/run-progressive-qualification"] +CMD ["--help"] diff --git a/containers/graphforge-progressive-qualification/run-qualification.py b/containers/graphforge-progressive-qualification/run-qualification.py new file mode 100644 index 00000000..813d5d98 --- /dev/null +++ b/containers/graphforge-progressive-qualification/run-qualification.py @@ -0,0 +1,82 @@ +#!/usr/local/bin/python3 +"""Fail-closed bootstrap for the progressive provider qualification runner.""" + +from __future__ import annotations + +import ctypes +import os +import stat +import sys + +WORK_ROOT = "/work" +RUN_UID = 10001 +RUN_GID = 10001 +PYTHON = "/opt/graphforge/benchmarks/.venv/bin/python" +MODULE = "graphforge_bench.progressive_provider_run" +PR_SET_NO_NEW_PRIVS = 38 + +# Do not forward the container's ambient environment. In particular, provider +# credentials and control-plane configuration must never reach the offline runner. +EXEC_ENV = { + "HOME": "/work", + "LANG": "C.UTF-8", + "PATH": "/opt/graphforge/benchmarks/.venv/bin:/usr/local/bin:/usr/bin:/bin", + "PYTHONDONTWRITEBYTECODE": "1", + "PYTHONPATH": "/opt/graphforge/benchmarks/harness", + "PYTHONUNBUFFERED": "1", +} + + +def refuse(message: str) -> None: + print(f"qualification bootstrap refused: {message}", file=sys.stderr) + raise SystemExit(64) + + +def validate_work_root() -> None: + try: + metadata = os.lstat(WORK_ROOT) + except OSError as error: + refuse(f"work root is unavailable: {error}") + if not stat.S_ISDIR(metadata.st_mode): + refuse("work root is not a directory") + if os.path.realpath(WORK_ROOT) != WORK_ROOT: + refuse("work root does not resolve to the exact mount path") + if not os.path.ismount(WORK_ROOT): + refuse("work root is not a mount point") + + +def enable_no_new_privileges() -> None: + libc = ctypes.CDLL(None, use_errno=True) + libc.prctl.argtypes = [ + ctypes.c_int, + ctypes.c_ulong, + ctypes.c_ulong, + ctypes.c_ulong, + ctypes.c_ulong, + ] + libc.prctl.restype = ctypes.c_int + if libc.prctl(PR_SET_NO_NEW_PRIVS, 1, 0, 0, 0) != 0: + error_number = ctypes.get_errno() + refuse(f"PR_SET_NO_NEW_PRIVS failed: {os.strerror(error_number)}") + + +def main() -> None: + if os.geteuid() != 0: + refuse("bootstrap must start as root") + validate_work_root() + + # Ownership changes only for the mount root itself, never its contents. + try: + os.chown(WORK_ROOT, RUN_UID, RUN_GID, follow_symlinks=False) + enable_no_new_privileges() + os.setgroups([]) + os.setgid(RUN_GID) + os.setuid(RUN_UID) + os.chdir(WORK_ROOT) + os.execve(PYTHON, [PYTHON, "-P", "-m", MODULE, *sys.argv[1:]], EXEC_ENV) + except OSError as error: + refuse(str(error)) + + +if __name__ == "__main__": + main() diff --git a/scripts/ci/test-progressive-qualification-image.py b/scripts/ci/test-progressive-qualification-image.py new file mode 100644 index 00000000..02ce38dc --- /dev/null +++ b/scripts/ci/test-progressive-qualification-image.py @@ -0,0 +1,451 @@ +#!/usr/bin/env python3 +"""Static, non-provider contract tests for the progressive qualification image.""" + +from __future__ import annotations + +import ast +import json +from pathlib import Path +import re +import shutil +import tempfile +import unittest + +import tomllib + +ROOT = Path(__file__).resolve().parents[2] +DOCKERFILE = Path("containers/graphforge-progressive-qualification/Dockerfile") +ENTRYPOINT = Path("containers/graphforge-progressive-qualification/run-qualification.py") +BENCHMARK_LOCK = Path("benchmarks/uv.lock") +DOCKERIGNORE = Path(".dockerignore") + +FORBIDDEN_RUNTIME = re.compile( + r"(?:^|\s)(?:EXPOSE|HEALTHCHECK)\b|" + r"\b(?:http\.server|uvicorn|gunicorn|flyctl|pulumi)\b|" + r"\b(?:ARG|ENV)\s+[A-Z0-9_]*(?:TOKEN|SECRET|PASSWORD|CREDENTIAL)", + re.IGNORECASE | re.MULTILINE, +) + + +class ContractError(AssertionError): + pass + + +def read(root: Path) -> str: + path = root / DOCKERFILE + if not path.is_file(): + raise ContractError(f"missing progressive qualification image: {DOCKERFILE}") + return path.read_text(encoding="utf-8") + + +def require(text: str, marker: str, message: str) -> None: + if marker not in text: + raise ContractError(message) + + +def rendered_manifest(dockerfile: str) -> dict[str, object]: + line = next( + ( + value.strip() + for value in dockerfile.splitlines() + if value.strip().startswith('"{\\"schema\\":\\"graphforge-progressive-provider-build/1') + ), + None, + ) + if line is None or not line.endswith('" \\'): + raise ContractError("build manifest JSON template is missing") + template = line.removesuffix(" \\").removeprefix('"').removesuffix('"').replace(r"\"", '"') + replacements = { + "${GRAPHFORGE_COMMIT}": "a" * 40, + "${source_tree_sha256}": "b" * 64, + "${gf_sha256}": "c" * 64, + "${certify_sha256}": "d" * 64, + "${generator_sha256}": "e" * 64, + "${benchexec_python_sha256}": "f" * 64, + } + for old, new in replacements.items(): + template = template.replace(old, new) + try: + document = json.loads(template) + except json.JSONDecodeError as error: + raise ContractError("build manifest template must be valid JSON") from error + return document + + +def bootstrap_constants(entrypoint: str) -> dict[str, object]: + """Return literal top-level bootstrap assignments for exact contract checks.""" + try: + tree = ast.parse(entrypoint) + except SyntaxError as error: + raise ContractError("startup boundary must be valid Python") from error + constants: dict[str, object] = {} + for statement in tree.body: + if not isinstance(statement, ast.Assign) or len(statement.targets) != 1: + continue + target = statement.targets[0] + if not isinstance(target, ast.Name): + continue + try: + constants[target.id] = ast.literal_eval(statement.value) + except (ValueError, TypeError): + continue + return constants + + +def validate_contract(root: Path) -> None: + dockerfile = read(root) + entrypoint = (root / ENTRYPOINT).read_text(encoding="utf-8") + stages = re.findall(r"^FROM\s+\S+\s+AS\s+(\S+)\s*$", dockerfile, re.MULTILINE) + if stages != ["rust-build", "python-deps", "qualification"]: + raise ContractError("image must use isolated Rust, locked Python, and runtime stages") + base_images = re.findall(r"^FROM\s+(\S+)\s+AS\s+\S+\s*$", dockerfile, re.MULTILINE) + if any(re.fullmatch(r"[^@]+@sha256:[0-9a-f]{64}", image) is None for image in base_images): + raise ContractError("every image stage must pin an immutable base digest") + + if dockerfile.count("ARG GRAPHFORGE_COMMIT") != 2: + raise ContractError("build and runtime stages must bind the same source identity") + require(dockerfile, "ARG TARGETARCH", "image build must receive the target architecture") + require(dockerfile, "ARG TARGETOS", "image build must receive the target operating system") + require( + dockerfile, + 'test "${TARGETOS}/${TARGETARCH}" = linux/amd64', + "image must fail closed off linux/amd64", + ) + require( + dockerfile, + "grep -Eq '^[0-9a-f]{40}$'", + "source identity must be a lowercase full Git object ID", + ) + require( + dockerfile, + 'LABEL org.opencontainers.image.revision="${GRAPHFORGE_COMMIT}"', + "runtime image must publish its source revision", + ) + require( + dockerfile, + "COPY --from=rust-build --chmod=0444 /graphforge-commit /opt/graphforge/commit", + "runtime image must carry a read-only source revision file", + ) + require( + dockerfile, + "tar --sort=name --mtime='UTC 1970-01-01' --owner=0 --group=0 --numeric-owner", + "build must deterministically archive the copied source tree", + ) + require( + dockerfile, + "sha256sum /tmp/source-tree.tar", + "build must hash the deterministic copied-source archive", + ) + + require( + dockerfile, + "cargo build --locked --release --package graphforge-cli --bin gf", + "Rust stage must build the locked gf CLI", + ) + require( + dockerfile, + "cargo build --manifest-path benchmarks/Cargo.toml --locked --release", + "Rust stage must use the locked benchmark workspace", + ) + for package in ( + "--package graphforge-benchmark-certify", + "--package graphforge-benchmark-graph500-generator", + ): + require(dockerfile, package, f"Rust stage must build {package.removeprefix('--package ')}") + + require( + dockerfile, + 'python -m pip install --no-cache-dir "uv==0.11.33"', + "Python dependency installer must use the pinned uv version", + ) + require( + dockerfile, + "COPY benchmarks/pyproject.toml benchmarks/uv.lock ./", + "Python stage must consume the benchmark lock", + ) + require( + dockerfile, + "uv sync --frozen --no-dev --no-install-project", + "Python dependencies must fail closed on lock drift", + ) + lock = tomllib.loads((root / BENCHMARK_LOCK).read_text(encoding="utf-8")) + benchexec = [package for package in lock["package"] if package["name"] == "benchexec"] + if len(benchexec) != 1 or not re.fullmatch(r"\d+\.\d+(?:\.\d+)?", benchexec[0]["version"]): + raise ContractError("benchmark lock must resolve exactly one pinned BenchExec package") + + ignored = (root / DOCKERIGNORE).read_text(encoding="utf-8").splitlines() + for pattern in (".git", "*.env", ".env*", "*.pem", "*.key"): + if pattern not in ignored: + raise ContractError(f"container build context must exclude {pattern}") + + for binary in ( + "/source/target/release/gf", + "/source/benchmarks/target/release/graphforge-benchmark-certify", + "/source/benchmarks/target/release/graphforge-benchmark-graph500-generator", + ): + require(dockerfile, binary, f"runtime must copy built binary {Path(binary).name}") + require( + dockerfile, + "> /opt/graphforge/build-manifest.json", + "runtime must contain the source and executable identity manifest", + ) + require( + dockerfile, + "chmod 0444 /opt/graphforge/build-manifest.json", + "build identity manifest must be read-only", + ) + manifest_fields = ( + r"\"schema\":\"graphforge-progressive-provider-build/1\"", + r"\"commit\":\"${GRAPHFORGE_COMMIT}\"", + r"\"source_tree_sha256\":\"${source_tree_sha256}\"", + r"\"executables\":{", + r"\"gf_sha256\":\"${gf_sha256}\"", + r"\"certify_sha256\":\"${certify_sha256}\"", + r"\"generator_executable_sha256\":\"${generator_sha256}\"", + r"\"benchexec_python_sha256\":\"${benchexec_python_sha256}\"", + ) + for field in manifest_fields: + require(dockerfile, field, "build manifest must bind its exact schema and identities") + manifest = rendered_manifest(dockerfile) + if set(manifest) != {"schema", "commit", "source_tree_sha256", "executables"}: + raise ContractError("build manifest must have the exact provider-build root shape") + if manifest["schema"] != "graphforge-progressive-provider-build/1": + raise ContractError("build manifest schema is not the runner contract") + if not re.fullmatch(r"[0-9a-f]{40}", str(manifest["commit"])): + raise ContractError("build manifest commit is not a full lowercase object ID") + if not re.fullmatch(r"[0-9a-f]{64}", str(manifest["source_tree_sha256"])): + raise ContractError("build manifest source tree identity is not SHA-256") + executables = manifest["executables"] + expected_executables = { + "gf_sha256", + "certify_sha256", + "generator_executable_sha256", + "benchexec_python_sha256", + } + if not isinstance(executables, dict) or set(executables) != expected_executables: + raise ContractError("build manifest must have the exact executable identity shape") + if any(not re.fullmatch(r"[0-9a-f]{64}", str(value)) for value in executables.values()): + raise ContractError("build manifest executable identities must be SHA-256") + for fixed_path in ( + "/usr/local/bin/gf", + "/usr/local/bin/graphforge-benchmark-certify", + "/usr/local/bin/graphforge-benchmark-graph500-generator", + "/opt/graphforge/benchmarks/.venv/bin/python", + ): + require(dockerfile, f"sha256sum {fixed_path}", f"manifest must hash {fixed_path}") + for asset in ("harness", "profiles", "schemas", "definitions"): + require( + dockerfile, + f"COPY benchmarks/{asset} /opt/graphforge/benchmarks/{asset}", + f"runtime must contain benchmark {asset}", + ) + + require(dockerfile, 'PYTHONPATH="/opt/graphforge/benchmarks/harness"', "harness must import") + require(dockerfile, "WORKDIR /", "startup must not enter the volume before validating it") + require( + dockerfile, + "COPY --chmod=0555 containers/graphforge-progressive-qualification/run-qualification.py", + "runtime must install its read-only startup boundary", + ) + require( + dockerfile, + 'ENTRYPOINT ["/usr/local/bin/run-progressive-qualification"]', + "image entrypoint must be the privilege-dropping startup boundary", + ) + require( + dockerfile, + 'CMD ["--help"]', + "default container invocation must not spend provider resources", + ) + + runtime = dockerfile.split(" AS qualification", 1)[-1] + if re.search(r"\b(?:apt-get|apt|useradd|adduser)\b", runtime): + raise ContractError("runtime stage may not resolve OS packages or create named users") + require(dockerfile, "chown 10001:10001 /work", "runtime must use the fixed numeric identity") + + constants = bootstrap_constants(entrypoint) + expected_environment = { + "HOME": "/work", + "LANG": "C.UTF-8", + "PATH": "/opt/graphforge/benchmarks/.venv/bin:/usr/local/bin:/usr/bin:/bin", + "PYTHONDONTWRITEBYTECODE": "1", + "PYTHONPATH": "/opt/graphforge/benchmarks/harness", + "PYTHONUNBUFFERED": "1", + } + if constants.get("WORK_ROOT") != "/work": + raise ContractError("startup must validate and enter exact /work") + if constants.get("RUN_UID") != 10001 or constants.get("RUN_GID") != 10001: + raise ContractError("startup must use the fixed numeric uid and gid") + if constants.get("EXEC_ENV") != expected_environment: + raise ContractError("startup executor environment must match the strict allowlist") + for marker in ( + "os.lstat(WORK_ROOT)", + "stat.S_ISDIR(metadata.st_mode)", + "os.path.realpath(WORK_ROOT) != WORK_ROOT", + "os.path.ismount(WORK_ROOT)", + ): + require(entrypoint, marker, "startup must validate the exact /work mount root") + require( + entrypoint, + "os.chown(WORK_ROOT, RUN_UID, RUN_GID, follow_symlinks=False)", + "startup must own only the mount root", + ) + if re.search(r"\b(?:chown|os\.chown)\b[^\n]*(?:recursive|-R)", entrypoint, re.IGNORECASE): + raise ContractError("startup may not recursively change volume ownership") + require(entrypoint, "os.chdir(WORK_ROOT)", "executor must run from the mounted work root") + require( + entrypoint, + "libc.prctl(PR_SET_NO_NEW_PRIVS, 1, 0, 0, 0)", + "startup must prohibit privilege acquisition", + ) + for marker in ("os.setgroups([])", "os.setgid(RUN_GID)", "os.setuid(RUN_UID)"): + require(entrypoint, marker, "startup must clear groups and drop numeric privileges") + require( + entrypoint, + 'os.execve(PYTHON, [PYTHON, "-P", "-m", MODULE, *sys.argv[1:]], EXEC_ENV)', + "startup must replace itself with the unprivileged executor", + ) + require( + entrypoint, + 'PYTHON = "/opt/graphforge/benchmarks/.venv/bin/python"', + "Python path must be fixed", + ) + require( + entrypoint, + 'MODULE = "graphforge_bench.progressive_provider_run"', + "module must be fixed", + ) + if re.search(r"\bos\.environ\b|\bos\.getenv\b", entrypoint): + raise ContractError("startup may not forward ambient environment variables") + for forbidden_name in ("TOKEN", "SECRET", "PASSWORD", "CREDENTIAL", "PULUMI", "FLY_"): + if forbidden_name in entrypoint: + raise ContractError(f"startup allowlist may not contain {forbidden_name}") + + combined_runtime = dockerfile + "\n" + entrypoint + match = FORBIDDEN_RUNTIME.search(combined_runtime) + if match: + raise ContractError( + f"qualification image exposes a service or credential input: {match.group(0)}" + ) + if re.search(r"\b(?:curl|wget|nc|ssh|socat)\b", entrypoint): + raise ContractError("startup boundary may not perform network operations") + + if re.search(r"\b(?:cargo|rustc|uv sync|pip install)\b", runtime): + raise ContractError("runtime stage may not contain build tooling or dependency resolution") + + +class ProgressiveQualificationImageTests(unittest.TestCase): + def test_repository_contract(self) -> None: + validate_contract(ROOT) + + def test_mutations_fail_closed(self) -> None: + with tempfile.TemporaryDirectory() as directory: + fixture = Path(directory) + for relative in (DOCKERFILE, ENTRYPOINT, BENCHMARK_LOCK, DOCKERIGNORE): + destination = fixture / relative + destination.parent.mkdir(parents=True, exist_ok=True) + shutil.copy2(ROOT / relative, destination) + dockerfile_mutations = { + "missing_commit_binding": ("ARG GRAPHFORGE_COMMIT", "ARG OMITTED_COMMIT"), + "malformed_commit_binding": ( + "grep -Eq '^[0-9a-f]{40}$'", + "grep -Eq '.*'", + ), + "wrong_platform": ('test "${TARGETOS}/${TARGETARCH}" = linux/amd64', "true"), + "mutable_base": ( + "@sha256:a339861ae23e9abb272cea45dfafde21760d2ce6577a70f8a926153677902663", + "", + ), + "mutable_python_dependencies": ("uv sync --frozen", "uv sync"), + "mutable_uv_installer": ('"uv==0.11.33"', '"uv"'), + "missing_generator": ( + "--package graphforge-benchmark-graph500-generator", + "--package omitted-generator", + ), + "missing_schema": ( + "COPY benchmarks/schemas /opt/graphforge/benchmarks/schemas", + "COPY benchmarks/omitted /opt/graphforge/benchmarks/omitted", + ), + "missing_source_tree_hash": ( + "sha256sum /tmp/source-tree.tar", + "echo unhashed-source-tree", + ), + "missing_manifest_source_identity": ( + r"\"source_tree_sha256\":\"${source_tree_sha256}\"", + r"\"source_tree_sha256\":\"omitted\"", + ), + "missing_manifest_binary_identity": ( + r"\"gf_sha256\":\"${gf_sha256}\"", + r"\"gf_sha256\":\"omitted\"", + ), + "extra_manifest_field": ( + r"\"executables\":{", + r"\"unexpected\":true,\"executables\":{", + ), + "writable_manifest": ( + "chmod 0444 /opt/graphforge/build-manifest.json", + "chmod 0644 /opt/graphforge/build-manifest.json", + ), + "public_service": ('CMD ["--help"]', "EXPOSE 8080"), + "credential_input": ('CMD ["--help"]', "ENV PROVIDER_TOKEN=unsafe"), + "floating_os_dependencies": ( + "RUN mkdir -p /opt/graphforge /work", + "RUN apt-get update && apt-get install -y util-linux\n" + "RUN mkdir -p /opt/graphforge /work", + ), + "named_runtime_user": ( + "RUN mkdir -p /opt/graphforge /work", + "RUN useradd benchexec\nRUN mkdir -p /opt/graphforge /work", + ), + } + for name, (old, new) in dockerfile_mutations.items(): + with self.subTest(name=name): + destination = fixture / DOCKERFILE + original = destination.read_text(encoding="utf-8") + self.assertIn(old, original) + destination.write_text(original.replace(old, new, 1), encoding="utf-8") + with self.assertRaises(ContractError): + validate_contract(fixture) + destination.write_text(original, encoding="utf-8") + + entrypoint_mutations = { + "plain_directory_work_root": ( + "os.path.ismount(WORK_ROOT)", + "os.path.isdir(WORK_ROOT)", + ), + "missing_privilege_drop": ("os.setuid(RUN_UID)", "pass"), + "recursive_chown": ( + "os.chown(WORK_ROOT, RUN_UID, RUN_GID, follow_symlinks=False)", + "os.chown(WORK_ROOT, RUN_UID, RUN_GID, recursive=True)", + ), + "network_behavior": ( + "os.chdir(WORK_ROOT)", + "os.system('wget https://provider.invalid')\n os.chdir(WORK_ROOT)", + ), + "ambient_environment": ( + '"HOME": "/work",', + '"HOME": os.environ["HOME"],', + ), + "expanded_environment": ( + '"HOME": "/work",', + '"UNREVIEWED": "value",\n "HOME": "/work",', + ), + "wrong_numeric_identity": ("RUN_UID = 10001", "RUN_UID = 10002"), + "unsafe_python_search_path": ( + '[PYTHON, "-P", "-m", MODULE, *sys.argv[1:]]', + '[PYTHON, "-m", MODULE, *sys.argv[1:]]', + ), + } + for name, (old, new) in entrypoint_mutations.items(): + with self.subTest(name=name): + destination = fixture / ENTRYPOINT + original = destination.read_text(encoding="utf-8") + self.assertIn(old, original) + destination.write_text(original.replace(old, new, 1), encoding="utf-8") + with self.assertRaises(ContractError): + validate_contract(fixture) + destination.write_text(original, encoding="utf-8") + + +if __name__ == "__main__": + unittest.main() From 2d5f0f4d03529133b26f19ed28be666d381896d4 Mon Sep 17 00:00:00 2001 From: David Spencer <1526975+DecisionNerd@users.noreply.github.com> Date: Sun, 30 Aug 2026 21:10:23 -0600 Subject: [PATCH 2/2] fix(bench): close provider runner review gaps --- benchmarks/Makefile | 4 +- benchmarks/README.md | 4 +- .../progressive_provider_run.py | 12 +++- .../graphforge_bench/progressive_run.py | 10 ++- .../progressive-provider-run-plan.json | 22 +++++++ .../progressive-provider-run-result.json | 2 +- .../tests/test_progressive_provider_plan.py | 31 ++++++++- .../tests/test_progressive_provider_run.py | 66 ++++++++++++++++++- benchmarks/tests/test_progressive_run.py | 15 +++++ .../run-qualification.py | 3 +- 10 files changed, 157 insertions(+), 12 deletions(-) diff --git a/benchmarks/Makefile b/benchmarks/Makefile index 549b2710..3e42442f 100644 --- a/benchmarks/Makefile +++ b/benchmarks/Makefile @@ -54,12 +54,12 @@ progressive-qualification-project-s20: install # Safe control-plane admission only; this never provisions a provider. progressive-provider-plan: install - test -n "$(COMMIT)" && test -n "$(MAXIMUM_SCALE)" && test -n "$(OUTPUT_DIR)" && test -n "$(PLAN_OUT)" && test -n "$(IMAGE_DIGEST)" + test -n "$(COMMIT)" && test -n "$(MAXIMUM_SCALE)" && test -n "$(OUTPUT_DIR)" && test -n "$(PLAN_OUT)" PYTHONPATH=$(CURDIR)/harness uv run --locked python -m graphforge_bench.qualification_operator \ plan-progressive -- \ --root "$(CURDIR)" --output-dir "$(OUTPUT_DIR)" --commit "$(COMMIT)" \ --maximum-scale "$(MAXIMUM_SCALE)" $(if $(PROVIDER_CAPACITY),--provider-capacity "$(PROVIDER_CAPACITY)",) \ - --image-digest "$(IMAGE_DIGEST)" \ + $(if $(IMAGE_DIGEST),--image-digest "$(IMAGE_DIGEST)",) \ --plan-out "$(PLAN_OUT)" # The registry owns the live command. Pulumi ESC supplies provider credentials; diff --git a/benchmarks/README.md b/benchmarks/README.md index a91f8ea2..b6a1006c 100644 --- a/benchmarks/README.md +++ b/benchmarks/README.md @@ -317,7 +317,9 @@ runner requires the image's read-only build manifest, matches the immutable `registry.fly.io/...@sha256:...` identity supplied by the admitted plan and provider transport, and revalidates the canonical profile, projection, source tree, native executables, and BenchExec identity before starting BenchExec. It -accepts only fixed in-image executable paths, an admitted plan and evidence +does not self-attest its OCI digest: the trusted provider transport must read +the provider-observed Machine image digest and supply that matching value. +It accepts only fixed in-image executable paths, an admitted plan and evidence directory below `/work`, no pre-existing files for the selected rung, and a real `/work` mount. The host must separately pass native Linux cgroups-v2 admission. The runner has no diff --git a/benchmarks/harness/graphforge_bench/progressive_provider_run.py b/benchmarks/harness/graphforge_bench/progressive_provider_run.py index 58b94289..476fe33c 100644 --- a/benchmarks/harness/graphforge_bench/progressive_provider_run.py +++ b/benchmarks/harness/graphforge_bench/progressive_provider_run.py @@ -239,7 +239,7 @@ def build_execution_plan( raise ProviderRunError("source tree digest must be a bare SHA-256 digest") scale, profile_path, profile = validate_admitted_plan(root, admitted_plan) if admitted_plan.get("image_digest") != image_digest: - raise ProviderRunError("running image does not match the admitted image digest") + raise ProviderRunError("transport-observed image digest contradicts admission") generator_identity = "sha256:" + _digest(root / "runners/graph500-generator/src/main.rs") if generator_identity != profile.get("generator", {}).get("identity"): raise ProviderRunError("generator source identity contradicts the provider profile") @@ -421,14 +421,20 @@ def run( _schema(root, "benchexec-run-evidence.json", benchexec) _schema(root, "certification-evidence.json", graphforge) _schema(root, "progressive-qualification-rung-evidence.json", rung) + except ProviderRunError as error: + failed = _result(plan, "failed", "ordinary_receipt_missing") + _schema(root, "progressive-provider-run-result.json", failed) + _write_json(result_path, failed) + raise ProviderRunError("ordinary_receipt_missing") from error + try: stored_plan, _ = _read_document(output_dir / f"s{scale}-plan.json") if dict(stored_plan) != dict(plan): raise ProviderRunError("stored execution plan changed during the run") except ProviderRunError as error: - failed = _result(plan, "failed", "ordinary_receipt_missing") + failed = _result(plan, "failed", "stored_plan_mismatch") _schema(root, "progressive-provider-run-result.json", failed) _write_json(result_path, failed) - raise ProviderRunError("ordinary_receipt_missing") from error + raise ProviderRunError("stored_plan_mismatch") from error artifact_paths = { "benchexec_sha256": output_dir / f"s{scale}-benchexec.json", "graphforge_sha256": output_dir / f"s{scale}-graphforge.json", diff --git a/benchmarks/harness/graphforge_bench/progressive_run.py b/benchmarks/harness/graphforge_bench/progressive_run.py index 1927a1ba..dd4f2a32 100644 --- a/benchmarks/harness/graphforge_bench/progressive_run.py +++ b/benchmarks/harness/graphforge_bench/progressive_run.py @@ -621,6 +621,14 @@ def ingest_benchexec_result( ) -> tuple[dict[str, Any], Mapping[str, Any], dict[str, Any]]: raw_output = stage / "raw" graphforge = _parse_graphforge_log(raw_output) + identities = plan.get("identities") + if not isinstance(identities, Mapping) or not isinstance(identities.get("profile_id"), str): + raise ControllerError("run plan profile identity is malformed") + planned_profile_id = str(identities["profile_id"]) + if graphforge.get("profile_id") != planned_profile_id: + raise ControllerError("certification profile identity contradicts the run plan") + if profile_id is not None and profile_id != planned_profile_id: + raise ControllerError("requested profile identity contradicts the run plan") raw = _parse_benchexec_xml(raw_output, correctness=graphforge.get("status") == "passed") limits = plan["limits"] benchexec = normalize_run( @@ -640,7 +648,7 @@ def ingest_benchexec_result( scale=scale, graphforge=graphforge, benchexec=benchexec, - profile_id=profile_id, + profile_id=planned_profile_id, source=source, ) return benchexec, graphforge, rung diff --git a/benchmarks/schemas/progressive-provider-run-plan.json b/benchmarks/schemas/progressive-provider-run-plan.json index 6598034b..dd9b9b75 100644 --- a/benchmarks/schemas/progressive-provider-run-plan.json +++ b/benchmarks/schemas/progressive-provider-run-plan.json @@ -46,5 +46,27 @@ }, "claim": {"const": "engineering_evidence_only"} }, + "allOf": [ + { + "if": {"properties": {"rung": {"const": "S20"}}, "required": ["rung"]}, + "then": {"properties": {"outputs": {"const": ["s20-plan.json", "s20-benchexec.json", "s20-graphforge.json", "s20-rung.json", "s20-result.json"]}}} + }, + { + "if": {"properties": {"rung": {"const": "S22"}}, "required": ["rung"]}, + "then": {"properties": {"outputs": {"const": ["s22-plan.json", "s22-benchexec.json", "s22-graphforge.json", "s22-rung.json", "s22-result.json"]}}} + }, + { + "if": {"properties": {"rung": {"const": "S24"}}, "required": ["rung"]}, + "then": {"properties": {"outputs": {"const": ["s24-plan.json", "s24-benchexec.json", "s24-graphforge.json", "s24-rung.json", "s24-result.json"]}}} + }, + { + "if": {"properties": {"rung": {"const": "S25"}}, "required": ["rung"]}, + "then": {"properties": {"outputs": {"const": ["s25-plan.json", "s25-benchexec.json", "s25-graphforge.json", "s25-rung.json", "s25-result.json"]}}} + }, + { + "if": {"properties": {"rung": {"const": "S26"}}, "required": ["rung"]}, + "then": {"properties": {"outputs": {"const": ["s26-plan.json", "s26-benchexec.json", "s26-graphforge.json", "s26-rung.json", "s26-result.json"]}}} + } + ], "$defs": {"digest": {"type": "string", "pattern": "^[0-9a-f]{64}$"}} } diff --git a/benchmarks/schemas/progressive-provider-run-result.json b/benchmarks/schemas/progressive-provider-run-result.json index 4eb34b3e..8e3e5143 100644 --- a/benchmarks/schemas/progressive-provider-run-result.json +++ b/benchmarks/schemas/progressive-provider-run-result.json @@ -9,7 +9,7 @@ "rung": {"enum": ["S20", "S22", "S24", "S25", "S26"]}, "status": {"enum": ["passed", "failed"]}, "failure": { - "enum": [null, "native_authority_unavailable", "staging_failed", "execution_boundary_failed", "benchexec_failed", "ordinary_receipt_missing"] + "enum": [null, "native_authority_unavailable", "staging_failed", "execution_boundary_failed", "benchexec_failed", "ordinary_receipt_missing", "stored_plan_mismatch"] }, "identities": { "type": "object", diff --git a/benchmarks/tests/test_progressive_provider_plan.py b/benchmarks/tests/test_progressive_provider_plan.py index 81f31bbf..dc218065 100644 --- a/benchmarks/tests/test_progressive_provider_plan.py +++ b/benchmarks/tests/test_progressive_provider_plan.py @@ -419,10 +419,39 @@ def test_provider_plan_has_offline_execution_authority(self) -> None: image_digest=IMAGE, ) require_execution_authority(plan) - refused = {**plan, "execution_authorized": False, "execution_refusal": "unavailable"} + refused = { + **plan, + "execution_authorized": False, + "execution_refusal": "provider_executor_unavailable", + } with self.assertRaisesRegex(ProviderPlanError, "authority is unavailable"): require_execution_authority(refused) + def test_make_target_only_forwards_an_available_image_digest(self) -> None: + variables = [ + f"COMMIT={COMMIT}", + "MAXIMUM_SCALE=19", + f"OUTPUT_DIR={self.output}", + f"PLAN_OUT={self.output / 'plan.json'}", + ] + without_image = subprocess.run( + ["make", "-n", "progressive-provider-plan", *variables], + cwd=ROOT, + capture_output=True, + check=True, + text=True, + ).stdout + self.assertNotIn("--image-digest", without_image) + + with_image = subprocess.run( + ["make", "-n", "progressive-provider-plan", *variables, f"IMAGE_DIGEST={IMAGE}"], + cwd=ROOT, + capture_output=True, + check=True, + text=True, + ).stdout + self.assertIn(f'--image-digest "{IMAGE}"', with_image) + def test_provider_plan_requires_one_immutable_admitted_image(self) -> None: with ( patch( diff --git a/benchmarks/tests/test_progressive_provider_run.py b/benchmarks/tests/test_progressive_provider_run.py index 8c8de3f0..b849c507 100644 --- a/benchmarks/tests/test_progressive_provider_run.py +++ b/benchmarks/tests/test_progressive_provider_run.py @@ -166,6 +166,12 @@ def test_plan_binds_admission_image_and_all_native_identities(self) -> None: hashlib.sha256((ROOT / "profiles/graph500/s20-provider.json").read_bytes()).hexdigest(), ) _schema(ROOT, "progressive-provider-run-plan.json", plan) + wrong_outputs = { + **plan, + "outputs": [name.replace("s20-", "s22-") for name in plan["outputs"]], + } + with self.assertRaisesRegex(ProviderRunError, "validation failed"): + _schema(ROOT, "progressive-provider-run-plan.json", wrong_outputs) def test_image_commit_attestation_works_without_git_metadata(self) -> None: image_root = self.base / "image" / "benchmarks" @@ -273,7 +279,7 @@ def test_admitted_image_must_match_the_running_image(self) -> None: "graphforge_bench.progressive_provider_run._benchexec_version", return_value="3.30", ), - self.assertRaisesRegex(ProviderRunError, "does not match"), + self.assertRaisesRegex(ProviderRunError, "contradicts admission"), ): build_execution_plan( root=ROOT, @@ -466,6 +472,58 @@ def test_missing_receipts_fail_closed(self) -> None: self.assertEqual(result["status"], "failed") self.assertEqual(result["failure"], "ordinary_receipt_missing") + def test_changed_stored_plan_has_distinct_typed_failure(self) -> None: + graphforge = graphforge_fixture(20) + graphforge["profile_id"] = "graph500-s20-provider" + benchexec = benchexec_fixture(graphforge) + rung = rung_fixture(20) + rung["profile_id"] = "graph500-s20-provider" + rung["source"] = "canonical_ladder" + for mutation in ("missing", "malformed", "mismatch"): + with self.subTest(mutation=mutation): + output = self.base / mutation + output.mkdir() + plan = self.execution_plan() + plan_path = output / "s20-plan.json" + plan_path.write_text(json.dumps(plan), encoding="utf-8") + + def change_plan( + *_: object, selected: str = mutation, selected_path: Path = plan_path + ) -> int: + if selected == "missing": + selected_path.unlink() + elif selected == "malformed": + selected_path.write_text("{", encoding="utf-8") + else: + selected_path.write_text(json.dumps({"changed": True}), encoding="utf-8") + return 0 + + with ( + patch( + "graphforge_bench.progressive_provider_run._benchexec_version", + return_value="3.30", + ), + patch( + "graphforge_bench.progressive_provider_run.ingest_benchexec_result", + return_value=(benchexec, graphforge, rung), + ), + self.assertRaisesRegex(ProviderRunError, "stored_plan_mismatch"), + ): + run( + root=ROOT, + output_dir=output, + plan=plan, + executables=self.executables, + expected_image_digest=IMAGE, + expected_admitted_plan_sha256="a" * 64, + expected_source_tree_sha256="b" * 64, + execution_boundary=change_plan, + authority_boundary=lambda: {"result": "passed"}, + ) + result = json.loads((output / "s20-result.json").read_text()) + self.assertEqual(result["failure"], "stored_plan_mismatch") + _schema(ROOT, "progressive-provider-run-result.json", result) + def test_success_emits_exactly_five_canonical_engineering_files(self) -> None: plan = self.execution_plan() self.output.mkdir() @@ -507,17 +565,21 @@ def test_success_emits_exactly_five_canonical_engineering_files(self) -> None: "s20-result.json", }, ) + stage = ingest.call_args.kwargs["stage"] self.assertEqual( ingest.call_args.kwargs, { "root": ROOT, - "stage": ingest.call_args.kwargs["stage"], + "stage": stage, "scale": 20, "plan": plan, "profile_id": "graph500-s20-provider", "source": "canonical_ladder", }, ) + self.assertEqual(stage.parent.parent, self.output) + self.assertTrue(stage.name.startswith("gf-progressive-")) + self.assertFalse(stage.exists()) result = json.loads((self.output / "s20-result.json").read_text()) self.assertEqual(result["claim"], "engineering_evidence_only") diff --git a/benchmarks/tests/test_progressive_run.py b/benchmarks/tests/test_progressive_run.py index b6c2d668..be101add 100644 --- a/benchmarks/tests/test_progressive_run.py +++ b/benchmarks/tests/test_progressive_run.py @@ -480,6 +480,21 @@ def test_exact_benchexec_xml_and_log_are_normalized_into_passed_bundle(self) -> self.assertEqual(normalized["authority"]["read_bytes"], 1024) self.assertEqual(rung["metrics"]["wall_seconds"], 2) + changed = {**gf, "profile_id": "graph500-s19-local"} + (raw / "run.log").write_text(json.dumps(changed) + "\n") + with self.assertRaisesRegex(ControllerError, "contradicts the run plan"): + ingest_benchexec_result(root=ROOT, stage=stage, scale=18, plan=plan) + + (raw / "run.log").write_text(json.dumps(gf) + "\n") + with self.assertRaisesRegex(ControllerError, "contradicts the run plan"): + ingest_benchexec_result( + root=ROOT, + stage=stage, + scale=18, + plan=plan, + profile_id="graph500-s19-local", + ) + def test_adjacent_passed_rungs_produce_schema_valid_s20_projection(self) -> None: for scale in (18, 19): (self.output / f"s{scale}-rung.json").write_text(json.dumps(passed_rung(scale))) diff --git a/containers/graphforge-progressive-qualification/run-qualification.py b/containers/graphforge-progressive-qualification/run-qualification.py index 813d5d98..0bf38006 100644 --- a/containers/graphforge-progressive-qualification/run-qualification.py +++ b/containers/graphforge-progressive-qualification/run-qualification.py @@ -7,6 +7,7 @@ import os import stat import sys +from typing import NoReturn WORK_ROOT = "/work" RUN_UID = 10001 @@ -27,7 +28,7 @@ } -def refuse(message: str) -> None: +def refuse(message: str) -> NoReturn: print(f"qualification bootstrap refused: {message}", file=sys.stderr) raise SystemExit(64)