diff --git a/.github/workflows/test.yml b/.github/workflows/test.yml index 0ade9efc..004edb00 100644 --- a/.github/workflows/test.yml +++ b/.github/workflows/test.yml @@ -1176,6 +1176,8 @@ jobs: working-directory: benchmarks run: | PYTHONPATH=harness uv run --locked python -m unittest \ + tests.test_progressive_esc \ + tests.test_progressive_fly_transport \ tests.test_progressive_provider_attempt \ tests.test_progressive_provider_plan \ tests.test_progressive_provider_run \ diff --git a/benchmarks/Makefile b/benchmarks/Makefile index 037dc33b..7ca0d5a7 100644 --- a/benchmarks/Makefile +++ b/benchmarks/Makefile @@ -1,4 +1,4 @@ -.PHONY: install smoke smoke-python smoke-rust fly-adapter-static progressive-provider-attempt-static local-admission progressive-qualification-list progressive-qualification-plan progressive-qualification-run progressive-qualification-project-s20 progressive-qualification-binaries progressive-provider-plan qualification-operator +.PHONY: install smoke smoke-python smoke-rust fly-adapter-static progressive-provider-attempt-static progressive-fly-transport-static local-admission progressive-qualification-list progressive-qualification-plan progressive-qualification-run progressive-qualification-project-s20 progressive-qualification-binaries progressive-provider-plan qualification-operator install: uv sync --locked @@ -24,6 +24,12 @@ progressive-provider-attempt-static: install PYTHONPATH=harness uv run --locked python -m unittest \ tests.test_progressive_provider_attempt +# Provider-free proof of the Fly command boundary and ESC secret capsule. +progressive-fly-transport-static: install + PYTHONPATH=harness uv run --locked python -m unittest \ + tests.test_progressive_fly_transport \ + tests.test_progressive_esc + local-admission: install PYTHONPATH=$(CURDIR)/harness uv run --locked reframe -C reframe/settings.py -c reframe/checks -n '^LocalBenchExecAdmission$$' -l | grep -F LocalBenchExecAdmission PYTHONPATH=$(CURDIR)/harness uv run --locked reframe -C reframe/settings.py -c reframe/checks -n '^LocalBenchExecAdmission$$' -r diff --git a/benchmarks/README.md b/benchmarks/README.md index f35dc847..43591db6 100644 --- a/benchmarks/README.md +++ b/benchmarks/README.md @@ -343,6 +343,22 @@ canonical five-file bundle; and persists an fsync-backed ownership ledger for cleanup-only recovery. Its transport is injected, so this proof makes no provider calls and spends nothing. +The production-shaped Fly boundary and isolated ESC input capsule are exercised +offline with: + +```bash +make -C benchmarks progressive-fly-transport-static +``` + +The transport accepts only an already-published immutable image, emits fixed +shell-free Fly commands through an injected boundary, applies the attempt +deadline to every operation, validates provider-observed Machine identity, +retrieves the result before the remaining canonical artifacts, and returns only +sanitized teardown counts. The ESC capsule consumes the fixed projected token +and spend-authorization variables once, removes them from the ambient process, +and constructs a minimal child environment with fresh credential state. Both +components remain import-only: these tests perform no provider operation. + Provider credentials belong to Pulumi ESC rather than GitHub workflow inputs or the caller's ambient shell. Live operator commands are rendered from `config/gate-registry.json` and run through the Python control plane: @@ -357,10 +373,12 @@ 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` is registered, but still fails before opening ESC. The -typed whole-attempt state machine, ownership-ledger recovery, and sanitized -teardown inventory now exist offline; the real Fly transport, ESC environment -binding, and live recovery gate remain deliberately unavailable. This is a -live capability boundary, not a GitHub-dispatch prerequisite. +typed whole-attempt state machine, production-shaped Fly command boundary, +isolated ESC input capsule, ownership-ledger recovery, and sanitized teardown +inventory now exist offline. Protected/versioned ESC configuration, immutable +image publication, an independently scheduled recovery owner or lease, and the +separately reviewed spend authorization remain prerequisites for live wiring. +This is a 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_esc.py b/benchmarks/harness/graphforge_bench/progressive_esc.py new file mode 100644 index 00000000..8c09ddcc --- /dev/null +++ b/benchmarks/harness/graphforge_bench/progressive_esc.py @@ -0,0 +1,228 @@ +"""Credential-isolated ESC inputs for the progressive Fly controller. + +This module is deliberately import-only. It does not invoke Pulumi, Fly, or +the progressive attempt controller and is not wired to an operator command. +""" + +from __future__ import annotations + +from collections.abc import Iterator, Mapping, MutableMapping +from dataclasses import dataclass, field +import os +from pathlib import Path +import tempfile +from typing import Any + +from graphforge_bench.progressive_provider_attempt import ( + SpendAuthorization, + parse_spend_authorization, +) + +FLY_TOKEN_ENV = "FLY_API_TOKEN" +SPEND_AUTHORIZATION_ENV = "GRAPHFORGE_PROGRESSIVE_SPEND_AUTHORIZATION" + +_CREDENTIAL_ALIASES = frozenset({"FLY_ACCESS_TOKEN"}) +_FORBIDDEN_OVERRIDES = frozenset( + { + "ALL_PROXY", + "DEBUG", + "FLY_DEBUG", + "FLY_LOG_LEVEL", + "HTTP_PROXY", + "HTTPS_PROXY", + "LOG_LEVEL", + "NO_PROXY", + "PULUMI_LOG_LEVEL", + "RUST_LOG", + } +) +_PROVIDER_PATH = "/usr/local/bin:/usr/bin:/bin" + + +class EscCapsuleError(ValueError): + """A sanitized refusal at the protected environment boundary.""" + + +class _Secret: + """Small non-printing holder for provider credential material.""" + + __slots__ = ("_value",) + + def __init__(self, value: str): + self._value = value + + def copy(self) -> str: + return self._value + + def clear(self) -> None: + self._value = "" + + def __repr__(self) -> str: + return "" + + +class _ProviderEnvironment(Mapping[str, str]): + """A subprocess-compatible mapping whose representation stays redacted.""" + + __slots__ = ("_fly_token", "_values") + + def __init__(self, fly_token: _Secret, values: Mapping[str, str]): + self._fly_token = fly_token + self._values = dict(values) + + def __getitem__(self, name: str) -> str: + if name == FLY_TOKEN_ENV: + return self._fly_token.copy() + return self._values[name] + + def __iter__(self) -> Iterator[str]: + yield FLY_TOKEN_ENV + yield from self._values + + def __len__(self) -> int: + return len(self._values) + 1 + + def __repr__(self) -> str: + return "ProviderEnvironment(FLY_API_TOKEN=, isolated_config=True)" + + +@dataclass(repr=False) +class ProgressiveEscCapsule: + """Validated ESC authority and an isolated environment for provider calls.""" + + _fly_token: _Secret + _authorization: SpendAuthorization | None + _temporary: tempfile.TemporaryDirectory[str] + home: Path + xdg_config_home: Path + _authorization_taken: bool = field(default=False, init=False) + _closed: bool = field(default=False, init=False) + _cleanup_complete: bool = field(default=False, init=False) + + def __repr__(self) -> str: + return "ProgressiveEscCapsule(fly_token=, authorization=)" + + def take_spend_authorization(self) -> SpendAuthorization: + """Return the parsed authority once, without retaining its encoded form.""" + if self._closed or self._authorization_taken or self._authorization is None: + raise EscCapsuleError("protected spend authorization is unavailable") + authorization = self._authorization + self._authorization = None + self._authorization_taken = True + return authorization + + def subprocess_environment(self) -> Mapping[str, str]: + """Build the complete, minimal environment for one Fly subprocess.""" + if self._closed: + raise EscCapsuleError("ESC capsule is closed") + return _ProviderEnvironment( + self._fly_token, + { + "HOME": str(self.home), + "LANG": "C.UTF-8", + "LC_ALL": "C.UTF-8", + "PATH": _PROVIDER_PATH, + "XDG_CONFIG_HOME": str(self.xdg_config_home), + }, + ) + + def close(self) -> None: + if self._cleanup_complete: + return + self._closed = True + self._fly_token.clear() + self._authorization = None + self._temporary.cleanup() + self._cleanup_complete = True + + def __enter__(self) -> ProgressiveEscCapsule: + if self._closed: + raise EscCapsuleError("ESC capsule is closed") + return self + + def __exit__(self, *_exc: Any) -> None: + self.close() + + +def _pop_projected_inputs( + environ: MutableMapping[str, str], +) -> tuple[str | None, str | None, bool]: + token: str | None = None + authorization: str | None = None + rejected = False + protected = {FLY_TOKEN_ENV, SPEND_AUTHORIZATION_ENV} + for name in list(environ): + normalized = name.upper() + if normalized not in protected | _CREDENTIAL_ALIASES: + continue + value = environ.pop(name) + if name == FLY_TOKEN_ENV and token is None: + token = value + elif name == SPEND_AUTHORIZATION_ENV and authorization is None: + authorization = value + else: + rejected = True + return token, authorization, rejected + + +def _reject_ambient_overrides(environ: MutableMapping[str, str]) -> None: + if any(name.upper() in _FORBIDDEN_OVERRIDES for name in environ): + raise EscCapsuleError("ambient credential or network override is forbidden") + + +def _validate_token(value: str | None) -> str: + if ( + not isinstance(value, str) + or not 1 <= len(value) <= 8192 + or value != value.strip() + or any(ord(character) < 0x20 or ord(character) == 0x7F for character in value) + ): + raise EscCapsuleError("projected Fly credential is unavailable or malformed") + return value + + +def _parse_authorization(value: str) -> SpendAuthorization | None: + """Keep parser exceptions and their protected input outside the public boundary.""" + try: + return parse_spend_authorization(value) + except Exception: + return None + + +def load_progressive_esc( + environ: MutableMapping[str, str] | None = None, +) -> ProgressiveEscCapsule: + """Consume exactly the two protected projections from the process environment.""" + source = os.environ if environ is None else environ + token_value, authorization_value, rejected_projection = _pop_projected_inputs(source) + try: + if rejected_projection: + raise EscCapsuleError("ambient credential or projected-input override is forbidden") + _reject_ambient_overrides(source) + token = _Secret(_validate_token(token_value)) + if not isinstance(authorization_value, str): + raise EscCapsuleError("protected spend authorization is unavailable") + authorization = _parse_authorization(authorization_value) + if authorization is None: + token.clear() + raise EscCapsuleError("protected spend authorization is invalid") + finally: + token_value = None + authorization_value = None + + try: + temporary = tempfile.TemporaryDirectory(prefix="graphforge-progressive-esc-") + except Exception: + token.clear() + raise + root = Path(temporary.name) + home = root / "home" + xdg_config_home = root / "xdg" + try: + home.mkdir(mode=0o700) + xdg_config_home.mkdir(mode=0o700) + except Exception: + token.clear() + temporary.cleanup() + raise + return ProgressiveEscCapsule(token, authorization, temporary, home, xdg_config_home) diff --git a/benchmarks/harness/graphforge_bench/progressive_fly_transport.py b/benchmarks/harness/graphforge_bench/progressive_fly_transport.py new file mode 100644 index 00000000..630e439a --- /dev/null +++ b/benchmarks/harness/graphforge_bench/progressive_fly_transport.py @@ -0,0 +1,842 @@ +"""Import-only Fly transport for progressive provider attempts. + +All provider I/O is delegated to an injected, shell-free boundary. This +module deliberately has no CLI and cannot open Pulumi ESC or start a live +qualification by itself. +""" + +from __future__ import annotations + +from collections.abc import Callable, Mapping, Sequence +from contextlib import suppress +from datetime import datetime, timedelta, timezone +import json +from pathlib import Path +import re +import shlex +import subprocess +import time +from typing import Any, Protocol +import urllib.error +import urllib.request + +from graphforge_bench.progressive_provider_attempt import ( + AttemptError, + AttemptInvocation, + ProvisionedAttempt, + SpendAuthorization, +) + +PROVIDER_RUNGS = (20, 22, 24, 25, 26) +IMAGE = re.compile(r"^registry\.fly\.io/[a-z0-9][a-z0-9._/-]*@sha256:[0-9a-f]{64}$") +OBSERVED_DIGEST = re.compile(r"^sha256:[0-9a-f]{64}$") +APP = re.compile(r"^gf-progressive-[0-9a-f]{32}$") +MACHINE_ID = re.compile(r"^[0-9a-f]{14}$") +VOLUME_ID = re.compile(r"^vol_[a-z0-9]+$") +CREATE_TIMEOUT_SECONDS = 300 +TRANSFER_TIMEOUT_SECONDS = 300 +TEARDOWN_TIMEOUT_SECONDS = 300 +TEARDOWN_POLL_ATTEMPTS = 6 +TEARDOWN_POLL_INTERVAL_SECONDS = 1 +REMOTE_OUTPUT_DIR = "/work/evidence" +API_ROOT = "https://api.machines.dev" +PROVIDER_ENVIRONMENT = frozenset( + {"FLY_API_TOKEN", "HOME", "LANG", "LC_ALL", "PATH", "XDG_CONFIG_HOME"} +) +ALLOWED_FLYCTL_COMMANDS = frozenset( + { + ("apps", "create"), + ("apps", "destroy"), + ("apps", "list"), + ("machine", "destroy"), + ("machine", "exec"), + ("machine", "list"), + ("machine", "run"), + ("secrets", "list"), + ("secrets", "unset"), + ("sftp", "get"), + ("sftp", "put"), + ("volumes", "create"), + ("volumes", "destroy"), + ("volumes", "list"), + } +) +FORBIDDEN_FLYCTL_ARGUMENTS = frozenset( + { + "--access-token", + "--build-depot", + "--build-nixpacks", + "--config", + "--debug", + "--dockerfile", + "--verbose", + "-c", + "-t", + } +) +SECRET_NAME = re.compile(r"^[A-Z_][A-Z0-9_]{0,127}$") +MACHINE_MEMORY_MB = { + **{f"shared-cpu-{cpus}x": cpus * 256 for cpus in (1, 2, 4, 6, 8)}, + **{f"performance-{cpus}x": cpus * 2048 for cpus in range(1, 17) if cpus % 2 == 0 or cpus == 1}, +} + + +class FlyTransportError(RuntimeError): + """A closed provider failure that contains no provider output.""" + + +class FlyBoundary(Protocol): + """Injectable boundary for argv execution and authoritative Machine state.""" + + def run( + self, argv: tuple[str, ...], *, timeout: int, check: bool = True + ) -> subprocess.CompletedProcess[str]: ... + + def json(self, argv: tuple[str, ...], *, timeout: int) -> Any: ... + + def api_json(self, path: str, *, timeout: int) -> Any: ... + + def machine_state(self, app: str, machine_id: str, *, timeout: int) -> Any: ... + + +class FlyctlMachineBoundary: + """Concrete shell-free flyctl and Machines API boundary. + + ``environment`` is expected to be the minimal mapping returned by the ESC + capsule. It is deliberately retained as a mapping rather than copied so + closing the capsule also clears the token visible to this boundary. + """ + + def __init__( + self, + environment: Mapping[str, str], + owner_app: str, + *, + cwd: Path | None = None, + urlopen: Callable[..., Any] | None = None, + ) -> None: + if set(environment) != PROVIDER_ENVIRONMENT: + raise FlyTransportError("provider environment is not isolated") + if APP.fullmatch(owner_app) is None: + raise FlyTransportError("provider app ownership is malformed") + self._environment = environment + self._owner_app = owner_app + self._cwd = cwd or Path.cwd() + self._urlopen = urlopen or urllib.request.urlopen + + def _env(self) -> dict[str, str]: + values = {name: self._environment[name] for name in PROVIDER_ENVIRONMENT} + if not values["FLY_API_TOKEN"]: + raise FlyTransportError("provider credential is unavailable") + return values + + def run( + self, argv: tuple[str, ...], *, timeout: int, check: bool = True + ) -> subprocess.CompletedProcess[str]: + if ( + len(argv) < 3 + or argv[0] != "flyctl" + or argv[1:3] not in ALLOWED_FLYCTL_COMMANDS + or self._has_forbidden_argument(argv[3:]) + or not self._targets_owner(argv) + ): + raise FlyTransportError("provider command is not allowed") + try: + completed = subprocess.run( + argv, + cwd=self._cwd, + env=self._env(), + shell=False, + check=False, + text=True, + capture_output=True, + timeout=timeout, + ) + except (OSError, subprocess.SubprocessError): + raise FlyTransportError("provider command failed") from None + if check and completed.returncode != 0: + raise FlyTransportError("provider command failed") + return completed + + @staticmethod + def _has_forbidden_argument(arguments: Sequence[str]) -> bool: + forbidden_long = {item for item in FORBIDDEN_FLYCTL_ARGUMENTS if item.startswith("--")} + forbidden_short = { + item for item in FORBIDDEN_FLYCTL_ARGUMENTS if item.startswith("-") and len(item) == 2 + } + return any( + argument in FORBIDDEN_FLYCTL_ARGUMENTS + or any(argument.startswith(f"{flag}=") for flag in forbidden_long) + or any(argument.startswith(flag) for flag in forbidden_short) + for argument in arguments + ) + + def _targets_owner(self, argv: tuple[str, ...]) -> bool: + command = argv[1:3] + arguments = argv[3:] + if command == ("apps", "list"): + return arguments == ("--json",) + if command in {("apps", "create"), ("apps", "destroy")}: + if not arguments or arguments[0] != self._owner_app: + return False + elif "--app" not in arguments and "-a" not in arguments: + return False + for index, argument in enumerate(arguments): + if argument in {"--app", "-a"}: + if index + 1 >= len(arguments) or arguments[index + 1] != self._owner_app: + return False + elif argument.startswith("--app=") or (argument.startswith("-a") and argument != "-a"): + return False + return True + + def json(self, argv: tuple[str, ...], *, timeout: int) -> Any: + try: + return json.loads(self.run(argv, timeout=timeout).stdout) + except (json.JSONDecodeError, UnicodeError): + raise FlyTransportError("provider JSON is malformed") from None + + def api_json(self, path: str, *, timeout: int) -> Any: + allowed_paths = { + f"/v1/apps/{self._owner_app}", + } + machine_path = re.fullmatch( + rf"/v1/apps/{re.escape(self._owner_app)}/machines/([0-9a-f]{{14}})", path + ) + volume_path = re.fullmatch( + rf"/v1/apps/{re.escape(self._owner_app)}/volumes/(vol_[a-z0-9]+)", path + ) + if path not in allowed_paths and machine_path is None and volume_path is None: + raise FlyTransportError("provider API path is not allowed") + request = urllib.request.Request( + API_ROOT + path, + headers={ + "Accept": "application/json", + "Authorization": f"Bearer {self._env()['FLY_API_TOKEN']}", + }, + ) + try: + with self._urlopen(request, timeout=timeout) as response: + payload = response.read(1_048_577) + if len(payload) > 1_048_576: + raise FlyTransportError("provider JSON is malformed") + return json.loads(payload) + except FlyTransportError: + raise + except (OSError, urllib.error.URLError, json.JSONDecodeError, UnicodeError): + raise FlyTransportError("provider API request failed") from None + + def machine_state(self, app: str, machine_id: str, *, timeout: int) -> Any: + if APP.fullmatch(app) is None or MACHINE_ID.fullmatch(machine_id) is None: + raise FlyTransportError("provider Machine identity is malformed") + return self.api_json(f"/v1/apps/{app}/machines/{machine_id}", timeout=timeout) + + +def _remaining_seconds( + deadline: datetime, + clock: Callable[[], datetime], + *, + maximum: int, +) -> int: + now = clock() + if deadline.tzinfo is None or now.tzinfo is None: + raise FlyTransportError("provider deadline is not timezone-aware") + remaining = (deadline.astimezone(timezone.utc) - now.astimezone(timezone.utc)).total_seconds() + if remaining <= 0: + raise FlyTransportError("provider deadline expired") + whole_seconds = int(remaining) + if whole_seconds < 1: + raise FlyTransportError("provider deadline expired") + return min(maximum, whole_seconds) + + +def _list(value: Any, label: str) -> list[Any]: + if not isinstance(value, list): + raise FlyTransportError(f"provider {label} inventory is malformed") + return value + + +def _app_names(value: Any) -> set[str]: + apps = _list(value, "app") + names: set[str] = set() + for item in apps: + name = item.get("Name") or item.get("name") if isinstance(item, Mapping) else None + if not isinstance(name, str): + raise FlyTransportError("provider app inventory is malformed") + names.add(name) + if len(names) != len(apps): + raise FlyTransportError("provider app inventory is malformed") + return names + + +def _single_volume_id(value: Any) -> str: + if isinstance(value, list) and len(value) == 1: + value = value[0] + volume_id = value.get("id") if isinstance(value, Mapping) else None + if not isinstance(volume_id, str) or VOLUME_ID.fullmatch(volume_id) is None: + raise FlyTransportError("created volume identity is malformed") + return volume_id + + +def _machine_id(value: Any, name: str) -> str: + machines = _list(value, "Machine") + if len(machines) != 1 or not isinstance(machines[0], Mapping): + raise FlyTransportError("provider Machine inventory is unexpected") + matches = [item for item in machines if isinstance(item, Mapping) and item.get("name") == name] + if len(matches) != 1: + raise FlyTransportError("created Machine identity is unavailable") + machine_id = matches[0].get("id") + if not isinstance(machine_id, str) or MACHINE_ID.fullmatch(machine_id) is None: + raise FlyTransportError("created Machine identity is malformed") + return machine_id + + +def _volume_items(value: Any) -> list[Mapping[str, Any]]: + volumes = _list(value, "volume") + if any(not isinstance(item, Mapping) for item in volumes): + raise FlyTransportError("provider volume inventory is malformed") + return volumes + + +def _secret_names(value: Any) -> list[str]: + secrets = _list(value, "secret") + names: list[str] = [] + for item in secrets: + name = item.get("Name") or item.get("name") if isinstance(item, Mapping) else None + if not isinstance(name, str) or SECRET_NAME.fullmatch(name) is None: + raise FlyTransportError("provider secret inventory is malformed") + names.append(name) + if len(set(names)) != len(names): + raise FlyTransportError("provider secret inventory is malformed") + return names + + +def _app_identity(value: Any, authorization: SpendAuthorization) -> None: + organization = value.get("organization") if isinstance(value, Mapping) else None + if ( + not isinstance(organization, Mapping) + or value.get("name") != authorization.app + or organization.get("slug") != authorization.organization + ): + raise FlyTransportError("provider app identity differs from authorization") + + +def _volume_identity( + value: Any, + authorization: SpendAuthorization, + *, + volume_id: str, + machine_id: str, +) -> None: + if not isinstance(value, Mapping): + raise FlyTransportError("provider volume state is malformed") + size_gb = value.get("size_gb", value.get("size_gb_total")) + if ( + value.get("id") != volume_id + or value.get("name") != f"{authorization.app}-data" + or value.get("region") != authorization.region + or size_gb != authorization.volume_gib + or value.get("auto_backup_enabled") is not False + or value.get("attached_machine_id") != machine_id + ): + raise FlyTransportError("provider volume state differs from authorization") + + +def _observed_image( + value: Any, + authorization: SpendAuthorization, + *, + volume_id: str, + machine_id: str, +) -> str: + if not isinstance(value, Mapping): + raise FlyTransportError("Machine state is malformed") + config = value.get("config") + image_ref = value.get("image_ref") + if not isinstance(config, Mapping) or not isinstance(image_ref, Mapping): + raise FlyTransportError("Machine state is incomplete") + guest = config.get("guest") + mounts = config.get("mounts") + restart = config.get("restart") + expected_kind, cpus_text = authorization.machine_class.rsplit("-", 1) + cpu_kind = "shared" if expected_kind == "shared-cpu" else "performance" + digest = image_ref.get("digest") + repository = authorization.image_digest.removeprefix("registry.fly.io/").rsplit("@", 1)[0] + metadata = config.get("metadata") + init = config.get("init") + expected_memory = MACHINE_MEMORY_MB.get(authorization.machine_class) + if ( + value.get("id") != machine_id + or value.get("name") != f"{authorization.app}-worker" + or value.get("state") != "started" + or value.get("region") != authorization.region + or not isinstance(value.get("private_ip"), str) + or not value["private_ip"].startswith("fdaa:") + or config.get("image") != authorization.image_digest + or config.get("auto_destroy") is not True + or not isinstance(restart, Mapping) + or restart.get("policy") != "no" + or config.get("services") not in (None, []) + or not isinstance(init, Mapping) + or init.get("entrypoint") not in (["/bin/sleep"], "/bin/sleep") + or init.get("cmd") not in (["infinity"], "infinity") + or not isinstance(guest, Mapping) + or guest.get("cpu_kind") != cpu_kind + or guest.get("cpus") != int(cpus_text.removesuffix("x")) + or guest.get("memory_mb") != expected_memory + or not isinstance(mounts, list) + or len(mounts) != 1 + or not isinstance(mounts[0], Mapping) + or mounts[0].get("path") != "/work" + or mounts[0].get("volume") != volume_id + or not isinstance(digest, str) + or OBSERVED_DIGEST.fullmatch(digest) is None + or image_ref.get("registry") != "registry.fly.io" + or image_ref.get("repository") != repository + or not isinstance(metadata, Mapping) + or metadata.get("graphforge_attempt_nonce") != authorization.attempt_nonce + or metadata.get("graphforge_commit") != authorization.commit + or metadata.get("graphforge_owner") != authorization.teardown_owner + or metadata.get("graphforge_machine_class") != authorization.machine_class + ): + raise FlyTransportError("Machine state differs from authorized resources") + return f"registry.fly.io/{repository}@{digest}" + + +class FlyProviderTransport: + """One-app, one-volume, one-Machine implementation of ProviderTransport.""" + + def __init__( + self, + boundary: FlyBoundary, + *, + clock: Callable[[], datetime] | None = None, + sleeper: Callable[[float], None] | None = None, + ) -> None: + self._boundary = boundary + self._clock = clock or (lambda: datetime.now(timezone.utc)) + self._sleeper = sleeper or time.sleep + self._app: str | None = None + self._machine_id: str | None = None + self._volume_id: str | None = None + self._authorization: SpendAuthorization | None = None + self._uploaded_rung: int | None = None + self._executed_rung: int | None = None + self._retrieved_results: set[int] = set() + + def _timeout(self, deadline: datetime, maximum: int) -> int: + return _remaining_seconds(deadline, self._clock, maximum=maximum) + + def _owned_machine(self) -> tuple[str, str]: + if self._app is None or self._machine_id is None: + raise FlyTransportError("provider Machine ownership is unavailable") + return self._app, self._machine_id + + def _validate_owned_state(self, deadline: datetime) -> str: + app, machine_id = self._owned_machine() + authorization = self._authorization + volume_id = self._volume_id + if authorization is None or volume_id is None: + raise FlyTransportError("provider ownership is unavailable") + _app_identity( + self._boundary.api_json( + f"/v1/apps/{app}", timeout=self._timeout(deadline, CREATE_TIMEOUT_SECONDS) + ), + authorization, + ) + machines = self._boundary.json( + ("flyctl", "machine", "list", "--app", app, "--json"), + timeout=self._timeout(deadline, CREATE_TIMEOUT_SECONDS), + ) + if _machine_id(machines, f"{app}-worker") != machine_id: + raise FlyTransportError("provider Machine inventory changed") + volumes = _volume_items( + self._boundary.json( + ("flyctl", "volumes", "list", "--app", app, "--json"), + timeout=self._timeout(deadline, CREATE_TIMEOUT_SECONDS), + ) + ) + if len(volumes) != 1 or volumes[0].get("id") != volume_id: + raise FlyTransportError("provider volume inventory is unexpected") + if _secret_names( + self._boundary.json( + ("flyctl", "secrets", "list", "--app", app, "--json"), + timeout=self._timeout(deadline, CREATE_TIMEOUT_SECONDS), + ) + ): + raise FlyTransportError("provider secret inventory is unexpected") + state = self._boundary.machine_state( + app, machine_id, timeout=self._timeout(deadline, CREATE_TIMEOUT_SECONDS) + ) + observed = _observed_image(state, authorization, volume_id=volume_id, machine_id=machine_id) + _volume_identity( + self._boundary.api_json( + f"/v1/apps/{app}/volumes/{volume_id}", + timeout=self._timeout(deadline, CREATE_TIMEOUT_SECONDS), + ), + authorization, + volume_id=volume_id, + machine_id=machine_id, + ) + return observed + + def provision( + self, + invocation: AttemptInvocation, + authorization: SpendAuthorization, + *, + deadline: datetime, + ) -> ProvisionedAttempt: + if ( + APP.fullmatch(authorization.app) is None + or IMAGE.fullmatch(authorization.image_digest) is None + or authorization.resource_limits + != {"apps": 1, "volumes": 1, "machines": 1, "image_builds": 0} + ): + raise FlyTransportError("provider authorization is incompatible") + self._authorization = authorization + if authorization.app in _app_names( + self._boundary.json( + ("flyctl", "apps", "list", "--json"), + timeout=self._timeout(deadline, CREATE_TIMEOUT_SECONDS), + ) + ): + raise FlyTransportError("authorized provider app already exists") + + self._app = authorization.app + self._boundary.run( + ( + "flyctl", + "apps", + "create", + authorization.app, + "--org", + authorization.organization, + "--json", + "--yes", + ), + timeout=self._timeout(deadline, CREATE_TIMEOUT_SECONDS), + ) + volume_name = f"{authorization.app}-data" + volume_id = _single_volume_id( + self._boundary.json( + ( + "flyctl", + "volumes", + "create", + volume_name, + "--app", + authorization.app, + "--region", + authorization.region, + "--size", + str(authorization.volume_gib), + "--count", + "1", + "--scheduled-snapshots=false", + "--json", + "--yes", + ), + timeout=self._timeout(deadline, CREATE_TIMEOUT_SECONDS), + ) + ) + self._volume_id = volume_id + machine_name = f"{authorization.app}-worker" + self._boundary.run( + ( + "flyctl", + "machine", + "run", + authorization.image_digest, + "infinity", + "--app", + authorization.app, + "--name", + machine_name, + "--metadata", + f"graphforge_attempt_nonce={authorization.attempt_nonce}", + "--metadata", + f"graphforge_commit={invocation.commit}", + "--metadata", + f"graphforge_owner={authorization.teardown_owner}", + "--metadata", + f"graphforge_machine_class={authorization.machine_class}", + "--region", + authorization.region, + "--vm-size", + authorization.machine_class, + "--volume", + f"{volume_id}:/work", + "--entrypoint", + "/bin/sleep", + "--restart", + "no", + "--autostop", + "off", + "--autostart=false", + "--rootfs-persist", + "never", + "--rm", + "--skip-dns-registration", + "--detach", + ), + timeout=self._timeout(deadline, CREATE_TIMEOUT_SECONDS), + ) + self._machine_id = _machine_id( + self._boundary.json( + ("flyctl", "machine", "list", "--app", authorization.app, "--json"), + timeout=self._timeout(deadline, CREATE_TIMEOUT_SECONDS), + ), + machine_name, + ) + observed = self._validate_owned_state(deadline) + return ProvisionedAttempt( + image_digest=observed, + resources={"machine_id": self._machine_id, "volume_id": volume_id}, + ) + + def upload_plan(self, *, rung: int, plan_path: Path, deadline: datetime) -> None: + app, machine_id = self._owned_machine() + if rung not in PROVIDER_RUNGS or not plan_path.is_file(): + raise FlyTransportError("admitted provider plan is unavailable") + remote = f"/work/s{rung}-admitted-plan.json" + self._boundary.run( + ( + "flyctl", + "sftp", + "put", + str(plan_path), + remote, + "--app", + app, + "--machine", + machine_id, + "--mode", + "0444", + "--quiet", + ), + timeout=self._timeout(deadline, TRANSFER_TIMEOUT_SECONDS), + ) + self._uploaded_rung = rung + + def execute_rung(self, *, rung: int, image_digest: str, deadline: datetime) -> int: + app, machine_id = self._owned_machine() + if ( + self._uploaded_rung != rung + or IMAGE.fullmatch(image_digest) is None + or self._validate_owned_state(deadline) != image_digest + ): + raise FlyTransportError("provider rung lacks an uploaded immutable plan") + remote = f"/work/s{rung}-admitted-plan.json" + command = shlex.join( + ( + "/usr/local/bin/run-progressive-qualification", + "--admitted-plan", + remote, + "--output-dir", + REMOTE_OUTPUT_DIR, + "--image-digest", + image_digest, + ) + ) + completed = self._boundary.run( + ( + "flyctl", + "machine", + "exec", + machine_id, + "--app", + app, + "--timeout", + str(self._timeout(deadline, 18_000)), + "--json", + command, + ), + timeout=self._timeout(deadline, 18_000), + check=False, + ) + self._executed_rung = rung + return completed.returncode + + def retrieve_result(self, *, rung: int, destination: Path, deadline: datetime) -> None: + if self._executed_rung != rung or destination.name != f"s{rung}-result.json": + raise FlyTransportError("provider result retrieval is not canonical") + self._retrieve( + rung=rung, + name=destination.name, + destination=destination, + deadline=deadline, + ) + self._retrieved_results.add(rung) + + def retrieve_success_artifacts( + self, + *, + rung: int, + names: Sequence[str], + destination: Path, + deadline: datetime, + ) -> None: + expected = tuple( + f"s{rung}-{suffix}.json" for suffix in ("plan", "benchexec", "graphforge", "rung") + ) + if ( + rung not in self._retrieved_results + or tuple(names) != expected + or not destination.is_dir() + ): + raise FlyTransportError("provider artifact retrieval is not canonical") + for name in expected: + self._retrieve( + rung=rung, + name=name, + destination=destination / name, + deadline=deadline, + ) + + def _retrieve( + self, + *, + rung: int, + name: str, + destination: Path, + deadline: datetime, + ) -> None: + app, machine_id = self._owned_machine() + allowed = { + f"s{rung}-{suffix}.json" + for suffix in ("plan", "benchexec", "graphforge", "rung", "result") + } + if name not in allowed: + raise FlyTransportError("provider evidence path is not allowed") + self._boundary.run( + ( + "flyctl", + "sftp", + "get", + f"{REMOTE_OUTPUT_DIR}/{name}", + str(destination), + "--app", + app, + "--machine", + machine_id, + "--quiet", + ), + timeout=self._timeout(deadline, TRANSFER_TIMEOUT_SECONDS), + ) + + def teardown(self, resources: Mapping[str, str]) -> Mapping[str, Any]: + app = resources.get("owner_app") + if not isinstance(app, str) or APP.fullmatch(app) is None: + raise FlyTransportError("provider teardown ownership is unavailable") + teardown_deadline = self._clock() + timedelta(seconds=TEARDOWN_TIMEOUT_SECONDS) + known_machine = resources.get("machine_id") + known_volume = resources.get("volume_id") + inventory_failure = False + unexpected_inventory = False + + def best_effort(argv: tuple[str, ...]) -> None: + with suppress(Exception): + self._boundary.run( + argv, + timeout=self._timeout(teardown_deadline, TEARDOWN_TIMEOUT_SECONDS), + check=False, + ) + + def app_exists() -> bool: + return app in _app_names( + self._boundary.json( + ("flyctl", "apps", "list", "--json"), + timeout=self._timeout(teardown_deadline, TEARDOWN_TIMEOUT_SECONDS), + ) + ) + + def inventory() -> tuple[list[str], list[str], list[str]]: + machines = _list( + self._boundary.json( + ("flyctl", "machine", "list", "--app", app, "--json"), + timeout=self._timeout(teardown_deadline, TEARDOWN_TIMEOUT_SECONDS), + ), + "Machine", + ) + machine_ids = [ + item.get("id") if isinstance(item, Mapping) else None for item in machines + ] + if any( + not isinstance(item, str) or MACHINE_ID.fullmatch(item) is None + for item in machine_ids + ) or len(set(machine_ids)) != len(machine_ids): + raise FlyTransportError("provider Machine inventory is malformed") + volumes = _volume_items( + self._boundary.json( + ("flyctl", "volumes", "list", "--app", app, "--json"), + timeout=self._timeout(teardown_deadline, TEARDOWN_TIMEOUT_SECONDS), + ) + ) + volume_ids = [item.get("id") for item in volumes] + if any( + not isinstance(item, str) or VOLUME_ID.fullmatch(item) is None + for item in volume_ids + ) or len(set(volume_ids)) != len(volume_ids): + raise FlyTransportError("provider volume inventory is malformed") + secrets = _secret_names( + self._boundary.json( + ("flyctl", "secrets", "list", "--app", app, "--json"), + timeout=self._timeout(teardown_deadline, TEARDOWN_TIMEOUT_SECONDS), + ) + ) + return machine_ids, volume_ids, secrets + + machine_ids: set[str] = set() + volume_ids: set[str] = set() + secrets: list[str] = [] + if isinstance(known_machine, str) and MACHINE_ID.fullmatch(known_machine): + machine_ids.add(known_machine) + if isinstance(known_volume, str) and VOLUME_ID.fullmatch(known_volume): + volume_ids.add(known_volume) + try: + exists = app_exists() + if exists: + observed_machines, observed_volumes, secrets = inventory() + machine_ids.update(observed_machines) + volume_ids.update(observed_volumes) + unexpected_inventory = ( + len(observed_machines) > 1 or len(observed_volumes) > 1 or bool(secrets) + ) + except Exception: + exists = True + inventory_failure = True + + if exists: + for machine_id in sorted(machine_ids): + best_effort(("flyctl", "machine", "destroy", "--app", app, "--force", machine_id)) + for volume_id in sorted(volume_ids): + best_effort(("flyctl", "volumes", "destroy", "--app", app, "--yes", volume_id)) + if secrets: + best_effort(("flyctl", "secrets", "unset", *secrets, "--app", app, "--yes")) + best_effort(("flyctl", "apps", "destroy", app, "--yes")) + + last = {"app_exists": False, "machines": 0, "volumes": 0, "secrets": 0} + try: + for attempt in range(TEARDOWN_POLL_ATTEMPTS): + if not app_exists(): + if inventory_failure or unexpected_inventory: + raise AttemptError( + "inventory_unavailable", "provider teardown inventory was anomalous" + ) + return last + remaining_machines, remaining_volumes, remaining_secrets = inventory() + last = { + "app_exists": True, + "machines": len(remaining_machines), + "volumes": len(remaining_volumes), + "secrets": len(remaining_secrets), + } + if attempt + 1 < TEARDOWN_POLL_ATTEMPTS: + self._sleeper(TEARDOWN_POLL_INTERVAL_SECONDS) + except AttemptError: + raise + except Exception as error: + raise AttemptError( + "inventory_unavailable", "provider teardown inventory is unavailable" + ) from error + return last diff --git a/benchmarks/tests/test_progressive_esc.py b/benchmarks/tests/test_progressive_esc.py new file mode 100644 index 00000000..6185fd98 --- /dev/null +++ b/benchmarks/tests/test_progressive_esc.py @@ -0,0 +1,229 @@ +from __future__ import annotations + +import hashlib +import json +import os +from pathlib import Path +import stat +import unittest +from unittest.mock import patch + +from graphforge_bench import progressive_esc +from graphforge_bench.progressive_esc import ( + FLY_TOKEN_ENV, + SPEND_AUTHORIZATION_ENV, + EscCapsuleError, + load_progressive_esc, +) + +COMMIT = "a" * 40 +NONCE = "b" * 32 +APP = f"gf-progressive-{NONCE}" +IMAGE = f"registry.fly.io/{APP}@sha256:" + "c" * 64 +TOKEN = "FlyV1 fixture-secret-token" + + +def authorization_document() -> dict[str, object]: + plan = { + "schema": "graphforge-progressive-provider-plan/1", + "status": "admitted", + "execution_authorized": True, + "execution_refusal": None, + "next_rung": "S20", + "image_digest": IMAGE, + } + plan_sha = hashlib.sha256( + (json.dumps(plan, indent=2, sort_keys=True) + "\n").encode() + ).hexdigest() + return { + "schema": "graphforge-progressive-spend-authorization/1", + "status": "authorized", + "provider": "fly", + "commit": COMMIT, + "admitted_plan_sha256": plan_sha, + "image_digest": IMAGE, + "organization": "fixture-org", + "region": "dfw", + "machine_class": "performance-4x", + "volume_gib": 500, + "rung": "S20", + "maximum_scale": 20, + "attempt_nonce": NONCE, + "app": APP, + "issued_at": "2026-06-01T00:00:00Z", + "expires_at": "2026-06-01T05:00:00Z", + "teardown_owner": "qualification-operator", + "maximum_machine_seconds": 18_000, + "resource_limits": {"apps": 1, "volumes": 1, "machines": 1, "image_builds": 0}, + "pricing": { + "currency": "USD", + "machine_microusd_per_hour": 1, + "volume_microusd_per_gib_hour": 1, + "transfer_allowance_microusd": 1, + "estimated_total_microusd": 2506, + "maximum_total_microusd": 3000, + }, + "claim": "spend_authorization_only", + } + + +def projected_environment(**extra: str) -> dict[str, str]: + return { + FLY_TOKEN_ENV: TOKEN, + SPEND_AUTHORIZATION_ENV: json.dumps(authorization_document()), + "UNRELATED_AMBIENT_VALUE": "must-not-propagate", + **extra, + } + + +class ProgressiveEscTests(unittest.TestCase): + def test_consumes_projections_once_and_redacts_representation(self) -> None: + environ = projected_environment() + with load_progressive_esc(environ) as capsule: + self.assertNotIn(FLY_TOKEN_ENV, environ) + self.assertNotIn(SPEND_AUTHORIZATION_ENV, environ) + encoded = repr(capsule) + self.assertNotIn(TOKEN, encoded) + self.assertNotIn(COMMIT, encoded) + self.assertEqual(capsule.take_spend_authorization().commit, COMMIT) + with self.assertRaises(EscCapsuleError): + capsule.take_spend_authorization() + with self.assertRaises(EscCapsuleError): + load_progressive_esc(environ) + + def test_provider_environment_is_minimal_and_uses_fresh_config(self) -> None: + with load_progressive_esc(projected_environment()) as capsule: + provider = capsule.subprocess_environment() + self.assertEqual( + set(provider), + {FLY_TOKEN_ENV, "HOME", "LANG", "LC_ALL", "PATH", "XDG_CONFIG_HOME"}, + ) + self.assertEqual(provider[FLY_TOKEN_ENV], TOKEN) + self.assertNotIn(TOKEN, repr(provider)) + self.assertNotIn(SPEND_AUTHORIZATION_ENV, provider) + self.assertNotIn("UNRELATED_AMBIENT_VALUE", provider) + for name in ("HOME", "XDG_CONFIG_HOME"): + path = Path(provider[name]) + self.assertTrue(path.is_dir()) + self.assertEqual(stat.S_IMODE(path.stat().st_mode), 0o700) + self.assertTrue(str(path).startswith(str(capsule.home.parent))) + self.assertFalse(capsule.home.parent.exists()) + self.assertEqual(provider[FLY_TOKEN_ENV], "") + with self.assertRaises(EscCapsuleError): + capsule.subprocess_environment() + + def test_missing_or_malformed_projections_are_scrubbed(self) -> None: + cases = ( + {}, + {FLY_TOKEN_ENV: TOKEN}, + {SPEND_AUTHORIZATION_ENV: json.dumps(authorization_document())}, + projected_environment(**{FLY_TOKEN_ENV: " token-with-whitespace "}), + projected_environment(**{SPEND_AUTHORIZATION_ENV: "{}"}), + ) + for environ in cases: + with self.subTest(environ=set(environ)), self.assertRaises(EscCapsuleError): + load_progressive_esc(environ) + self.assertNotIn(FLY_TOKEN_ENV, environ) + self.assertNotIn(SPEND_AUTHORIZATION_ENV, environ) + + def test_rejects_aliases_and_network_or_logging_overrides(self) -> None: + for name in ( + "FLY_ACCESS_TOKEN", + "fly_access_token", + "HTTP_PROXY", + "https_proxy", + "ALL_PROXY", + "NO_PROXY", + "FLY_DEBUG", + "FLY_LOG_LEVEL", + "LOG_LEVEL", + "PULUMI_LOG_LEVEL", + "RUST_LOG", + "DEBUG", + ): + environ = projected_environment(**{name: "secret-alias"}) + with self.subTest(name=name), self.assertRaisesRegex(EscCapsuleError, "override"): + load_progressive_esc(environ) + self.assertNotIn(FLY_TOKEN_ENV, environ) + self.assertNotIn(SPEND_AUTHORIZATION_ENV, environ) + if name.upper() == "FLY_ACCESS_TOKEN": + self.assertNotIn(name, environ) + + def test_rejects_and_scrubs_all_case_variants_and_duplicate_aliases(self) -> None: + cases = ( + {"fly_api_token": "lower-token"}, + {"graphforge_progressive_spend_authorization": "lower-authorization"}, + {"FLY_ACCESS_TOKEN": "first-alias", "fly_access_token": "second-alias"}, + ) + protected = { + FLY_TOKEN_ENV, + SPEND_AUTHORIZATION_ENV, + "FLY_ACCESS_TOKEN", + } + for extras in cases: + environ = projected_environment(**extras) + with self.subTest(extras=set(extras)), self.assertRaises(EscCapsuleError): + load_progressive_esc(environ) + self.assertFalse(any(name.upper() in protected for name in environ)) + + def test_invalid_authorization_is_not_retained_by_exception_chain(self) -> None: + authorization_canary = "protected-authorization-canary" + token_canary = "protected-token-canary" + environ = { + FLY_TOKEN_ENV: token_canary, + SPEND_AUTHORIZATION_ENV: f'{{"canary":"{authorization_canary}", BROKEN', + } + with self.assertRaises(EscCapsuleError) as raised: + load_progressive_esc(environ) + self.assertIsNone(raised.exception.__cause__) + self.assertIsNone(raised.exception.__context__) + self.assertNotIn(authorization_canary, repr(raised.exception)) + self.assertNotIn(token_canary, repr(raised.exception)) + + def test_cleanup_failure_disables_capsule_and_remains_retryable(self) -> None: + capsule = load_progressive_esc(projected_environment()) + root = capsule.home.parent + cleanup = capsule._temporary.cleanup + attempts = 0 + + def flaky_cleanup() -> None: + nonlocal attempts + attempts += 1 + if attempts == 1: + raise OSError("fixture cleanup failure") + cleanup() + + with patch.object(capsule._temporary, "cleanup", side_effect=flaky_cleanup): + with self.assertRaises(OSError): + capsule.close() + with self.assertRaises(EscCapsuleError): + capsule.subprocess_environment() + with self.assertRaises(EscCapsuleError): + capsule.take_spend_authorization() + self.assertTrue(root.exists()) + capsule.close() + self.assertEqual(attempts, 2) + self.assertFalse(root.exists()) + + def test_module_has_no_command_execution_surface(self) -> None: + source = Path(progressive_esc.__file__).read_text(encoding="utf-8") + self.assertNotIn("import subprocess", source) + self.assertNotIn("flyctl", source) + self.assertNotIn("def main(", source) + + def test_default_loader_consumes_real_process_environment(self) -> None: + before = dict(os.environ) + try: + os.environ.clear() + os.environ.update(projected_environment()) + with load_progressive_esc() as capsule: + self.assertEqual(capsule.take_spend_authorization().app, APP) + self.assertNotIn(FLY_TOKEN_ENV, os.environ) + self.assertNotIn(SPEND_AUTHORIZATION_ENV, os.environ) + finally: + os.environ.clear() + os.environ.update(before) + + +if __name__ == "__main__": + unittest.main() diff --git a/benchmarks/tests/test_progressive_fly_transport.py b/benchmarks/tests/test_progressive_fly_transport.py new file mode 100644 index 00000000..8216052a --- /dev/null +++ b/benchmarks/tests/test_progressive_fly_transport.py @@ -0,0 +1,519 @@ +from __future__ import annotations + +from datetime import datetime, timedelta, timezone +import json +from pathlib import Path +import subprocess +import tempfile +import unittest +from unittest import mock + +from graphforge_bench.progressive_fly_transport import ( + FlyctlMachineBoundary, + FlyProviderTransport, + FlyTransportError, +) +from graphforge_bench.progressive_provider_attempt import AttemptError, AttemptInvocation, execute +from tests.test_progressive_provider_attempt import APP, IMAGE, ROOT, authorization, planner + +NOW = datetime(2026, 6, 1, tzinfo=timezone.utc) +MACHINE_ID = "abcdef01234567" +VOLUME_ID = "vol_fixture123" + + +class FakeBoundary: + def __init__(self) -> None: + self.calls: list[tuple[object, ...]] = [] + self.app_exists = False + self.keep_app = False + self.destroy_raises = False + self.inventory_raises = False + self.observed_digest = "sha256:" + "1" * 64 + self.observed_repository = APP + self.fail_prefix: tuple[str, ...] | None = None + self.secrets: list[dict[str, str]] = [] + + def run( + self, argv: tuple[str, ...], *, timeout: int, check: bool = True + ) -> subprocess.CompletedProcess[str]: + self.calls.append(("run", argv, timeout, check)) + if argv[:3] == ("flyctl", "apps", "create"): + self.app_exists = True + if self.fail_prefix is not None and argv[: len(self.fail_prefix)] == self.fail_prefix: + raise OSError("secret-canary provider failure") + if len(argv) >= 3 and argv[:3] == ("flyctl", "apps", "destroy"): + if self.destroy_raises: + raise RuntimeError("sensitive provider diagnostic") + if not self.keep_app: + self.app_exists = False + if argv[:3] == ("flyctl", "sftp", "get"): + Path(argv[4]).write_text("{}", encoding="utf-8") + return subprocess.CompletedProcess(argv, 0, "", "") + + def json(self, argv: tuple[str, ...], *, timeout: int) -> object: + self.calls.append(("json", argv, timeout)) + if self.fail_prefix is not None and argv[: len(self.fail_prefix)] == self.fail_prefix: + raise OSError("secret-canary provider failure") + if self.inventory_raises and argv[:4] == ("flyctl", "apps", "list", "--json"): + raise OSError("sensitive provider diagnostic") + if argv[:4] == ("flyctl", "apps", "list", "--json"): + return [{"Name": APP}] if self.app_exists else [] + if argv[:3] == ("flyctl", "volumes", "create"): + return {"id": VOLUME_ID} + if argv[:3] == ("flyctl", "machine", "list"): + return [{"id": MACHINE_ID, "name": f"{APP}-worker"}] + if argv[:3] == ("flyctl", "volumes", "list"): + return [{"id": VOLUME_ID}] + if argv[:3] == ("flyctl", "secrets", "list"): + return self.secrets + raise AssertionError(argv) + + def machine_state(self, app: str, machine_id: str, *, timeout: int) -> object: + self.calls.append(("machine_state", app, machine_id, timeout)) + return { + "id": MACHINE_ID, + "name": f"{APP}-worker", + "state": "started", + "region": "dfw", + "private_ip": "fdaa::1", + "config": { + "image": IMAGE, + "auto_destroy": True, + "init": {"entrypoint": ["/bin/sleep"], "cmd": ["infinity"]}, + "restart": {"policy": "no"}, + "services": [], + "guest": {"cpu_kind": "performance", "cpus": 4, "memory_mb": 8192}, + "mounts": [{"path": "/work", "volume": VOLUME_ID}], + "metadata": { + "graphforge_attempt_nonce": self.authorization.attempt_nonce, + "graphforge_commit": self.authorization.commit, + "graphforge_owner": self.authorization.teardown_owner, + "graphforge_machine_class": self.authorization.machine_class, + }, + }, + "image_ref": { + "registry": "registry.fly.io", + "repository": self.observed_repository, + "digest": self.observed_digest, + }, + } + + def api_json(self, path: str, *, timeout: int) -> object: + self.calls.append(("api_json", path, timeout)) + if path == f"/v1/apps/{APP}": + return { + "name": APP, + "organization": {"slug": self.authorization.organization}, + } + if path == f"/v1/apps/{APP}/volumes/{VOLUME_ID}": + return { + "id": VOLUME_ID, + "name": f"{APP}-data", + "region": self.authorization.region, + "size_gb": self.authorization.volume_gib, + "auto_backup_enabled": False, + "attached_machine_id": MACHINE_ID, + } + raise AssertionError(path) + + +class ProgressiveFlyTransportTests(unittest.TestCase): + def setUp(self) -> None: + self.temporary = tempfile.TemporaryDirectory() + self.base = Path(self.temporary.name) + self.boundary = FakeBoundary() + self.auth = authorization(20) + self.boundary.authorization = self.auth + self.transport = FlyProviderTransport( + self.boundary, clock=lambda: NOW, sleeper=lambda _seconds: None + ) + self.invocation = AttemptInvocation( + ROOT, + self.base / "evidence", + self.base / "ledger.json", + self.auth.commit, + ) + self.deadline = NOW + timedelta(hours=1) + + def tearDown(self) -> None: + self.temporary.cleanup() + + def provision(self) -> None: + result = self.transport.provision(self.invocation, self.auth, deadline=self.deadline) + self.assertEqual(result.image_digest, IMAGE) + self.assertEqual(result.resources, {"machine_id": MACHINE_ID, "volume_id": VOLUME_ID}) + + def test_provision_is_zero_build_private_fixed_and_provider_observed(self) -> None: + self.provision() + commands = [call[1] for call in self.boundary.calls if call[0] == "run"] + rendered = "\n".join(" ".join(command) for command in commands) + self.assertNotIn(" deploy ", f" {rendered} ") + self.assertNotIn("build", rendered) + machine = next( + command for command in commands if command[:3] == ("flyctl", "machine", "run") + ) + self.assertEqual( + machine, + ( + "flyctl", + "machine", + "run", + IMAGE, + "infinity", + "--app", + APP, + "--name", + f"{APP}-worker", + "--metadata", + f"graphforge_attempt_nonce={self.auth.attempt_nonce}", + "--metadata", + f"graphforge_commit={self.auth.commit}", + "--metadata", + f"graphforge_owner={self.auth.teardown_owner}", + "--metadata", + f"graphforge_machine_class={self.auth.machine_class}", + "--region", + "dfw", + "--vm-size", + "performance-4x", + "--volume", + f"{VOLUME_ID}:/work", + "--entrypoint", + "/bin/sleep", + "--restart", + "no", + "--autostop", + "off", + "--autostart=false", + "--rootfs-persist", + "never", + "--rm", + "--skip-dns-registration", + "--detach", + ), + ) + self.assertTrue(any(call[0] == "machine_state" for call in self.boundary.calls)) + + def test_machine_readback_digest_is_not_synthesized_from_authorization(self) -> None: + self.boundary.observed_digest = "sha256:" + "2" * 64 + result = self.transport.provision(self.invocation, self.auth, deadline=self.deadline) + self.assertEqual( + result.image_digest, + f"registry.fly.io/{APP}@sha256:" + "2" * 64, + ) + self.assertNotEqual(result.image_digest, self.auth.image_digest) + + def test_wrong_provider_repository_is_refused_even_with_authorized_digest(self) -> None: + self.boundary.observed_repository = "another-app" + with self.assertRaisesRegex(FlyTransportError, "differs"): + self.transport.provision(self.invocation, self.auth, deadline=self.deadline) + + def test_existing_app_refuses_before_mutation(self) -> None: + self.boundary.app_exists = True + with self.assertRaisesRegex(FlyTransportError, "already exists"): + self.transport.provision(self.invocation, self.auth, deadline=self.deadline) + self.assertFalse(any(call[0] == "run" for call in self.boundary.calls)) + + def test_expired_deadline_refuses_before_provider_call(self) -> None: + with self.assertRaisesRegex(FlyTransportError, "expired"): + self.transport.provision( + self.invocation, self.auth, deadline=NOW - timedelta(seconds=1) + ) + self.assertEqual(self.boundary.calls, []) + + def test_upload_execute_and_result_first_retrieval_are_canonical(self) -> None: + self.provision() + plan = self.base / "control-plan.json" + plan.write_text("{}", encoding="utf-8") + self.transport.upload_plan(rung=20, plan_path=plan, deadline=self.deadline) + with tempfile.TemporaryDirectory(dir=self.base) as directory: + stage = Path(directory) + names = tuple( + f"s20-{suffix}.json" for suffix in ("plan", "benchexec", "graphforge", "rung") + ) + with self.assertRaisesRegex(FlyTransportError, "canonical"): + self.transport.retrieve_success_artifacts( + rung=20, + names=names, + destination=stage, + deadline=self.deadline, + ) + self.assertEqual( + self.transport.execute_rung(rung=20, image_digest=IMAGE, deadline=self.deadline), + 0, + ) + self.transport.retrieve_result( + rung=20, + destination=stage / "s20-result.json", + deadline=self.deadline, + ) + self.transport.retrieve_success_artifacts( + rung=20, + names=names, + destination=stage, + deadline=self.deadline, + ) + transfers = [ + call[1] for call in self.boundary.calls if call[0] == "run" and call[1][1] == "sftp" + ] + self.assertEqual(transfers[0][2], "put") + self.assertEqual(transfers[1][3], "/work/evidence/s20-result.json") + self.assertEqual( + [command[3] for command in transfers[2:]], + [f"/work/evidence/{name}" for name in names], + ) + executed = next( + call + for call in self.boundary.calls + if call[0] == "run" and call[1][1:3] == ("machine", "exec") + ) + self.assertFalse(executed[3]) + self.assertIn("/usr/local/bin/run-progressive-qualification", executed[1][-1]) + self.assertNotIn("FLY_API_TOKEN", " ".join(executed[1])) + + def test_teardown_is_best_effort_then_independently_observed(self) -> None: + self.boundary.app_exists = True + self.boundary.destroy_raises = True + self.boundary.keep_app = True + observed = self.transport.teardown( + { + "owner_app": APP, + "machine_id": MACHINE_ID, + "volume_id": VOLUME_ID, + } + ) + self.assertEqual(observed, {"app_exists": True, "machines": 1, "volumes": 1, "secrets": 0}) + operations = [call[1][:3] for call in self.boundary.calls if call[0] == "run"] + self.assertEqual( + operations, + [ + ("flyctl", "machine", "destroy"), + ("flyctl", "volumes", "destroy"), + ("flyctl", "apps", "destroy"), + ], + ) + + def test_owner_only_crash_recovery_destroys_app(self) -> None: + self.boundary.app_exists = True + observed = self.transport.teardown({"owner_app": APP}) + self.assertEqual(observed, {"app_exists": False, "machines": 0, "volumes": 0, "secrets": 0}) + + def test_inventory_failure_is_typed_and_sanitized(self) -> None: + self.boundary.app_exists = True + self.boundary.inventory_raises = True + with self.assertRaises(AttemptError) as raised: + self.transport.teardown({"owner_app": APP}) + self.assertEqual(raised.exception.failure, "inventory_unavailable") + self.assertNotIn("sensitive", str(raised.exception)) + + def test_subsecond_deadline_never_rounds_up_into_provider_call(self) -> None: + with self.assertRaisesRegex(FlyTransportError, "expired"): + self.transport.provision( + self.invocation, + self.auth, + deadline=NOW + timedelta(milliseconds=999), + ) + self.assertEqual(self.boundary.calls, []) + + def test_machine_identity_is_revalidated_immediately_before_execution(self) -> None: + self.provision() + plan = self.base / "control-plan.json" + plan.write_text("{}", encoding="utf-8") + self.transport.upload_plan(rung=20, plan_path=plan, deadline=self.deadline) + self.boundary.observed_digest = "sha256:" + "2" * 64 + before = len(self.boundary.calls) + with self.assertRaisesRegex(FlyTransportError, "uploaded immutable plan"): + self.transport.execute_rung(rung=20, image_digest=IMAGE, deadline=self.deadline) + self.assertTrue(any(call[0] == "machine_state" for call in self.boundary.calls[before:])) + self.assertFalse( + any( + call[0] == "run" and call[1][1:3] == ("machine", "exec") + for call in self.boundary.calls[before:] + ) + ) + + def test_ambiguous_mutation_faults_always_reach_inventory_teardown(self) -> None: + for index, prefix in enumerate( + ( + ("flyctl", "apps", "create"), + ("flyctl", "volumes", "create"), + ("flyctl", "machine", "run"), + ) + ): + with self.subTest(prefix=prefix): + boundary = FakeBoundary() + auth = authorization(20) + boundary.authorization = auth + boundary.fail_prefix = prefix + transport = FlyProviderTransport( + boundary, clock=lambda: NOW, sleeper=lambda _seconds: None + ) + case = self.base / f"fault-{index}" + outcome = execute( + AttemptInvocation( + ROOT, + case / "evidence", + case / "ledger.json", + auth.commit, + ), + auth, + transport=transport, + planner=planner, + prefix_reader=lambda *_args, **_kwargs: [ + {"scale": 18}, + {"scale": 19}, + ], + now=NOW, + clock=lambda: NOW, + ) + self.assertEqual(outcome.failure, "provision_failed") + self.assertTrue( + any( + call[0] == "run" and call[1][:3] == ("flyctl", "apps", "destroy") + for call in boundary.calls + ) + ) + + def test_owner_only_teardown_discovers_and_removes_every_resource(self) -> None: + self.boundary.app_exists = True + self.boundary.json = mock.Mock(wraps=self.boundary.json) + observed = self.transport.teardown({"owner_app": APP}) + self.assertEqual(observed, {"app_exists": False, "machines": 0, "volumes": 0, "secrets": 0}) + operations = [call[1][:3] for call in self.boundary.calls if call[0] == "run"] + self.assertEqual( + operations, + [ + ("flyctl", "machine", "destroy"), + ("flyctl", "volumes", "destroy"), + ("flyctl", "apps", "destroy"), + ], + ) + + def test_unexpected_secret_is_removed_but_reported_as_sanitized_anomaly(self) -> None: + self.boundary.app_exists = True + self.boundary.secrets = [{"Name": "SECRET_CANARY"}] + with self.assertRaises(AttemptError) as raised: + self.transport.teardown({"owner_app": APP}) + self.assertEqual(raised.exception.failure, "inventory_unavailable") + self.assertNotIn("SECRET_CANARY", str(raised.exception)) + unset = next( + call[1] + for call in self.boundary.calls + if call[0] == "run" and call[1][:3] == ("flyctl", "secrets", "unset") + ) + self.assertIn("SECRET_CANARY", unset) + self.assertTrue( + any( + call[0] == "run" and call[1][:3] == ("flyctl", "apps", "destroy") + for call in self.boundary.calls + ) + ) + + def test_concrete_boundary_uses_exact_environment_and_never_a_shell(self) -> None: + token = "FlyV1 secret-canary" + environment = { + "FLY_API_TOKEN": token, + "HOME": "/tmp/fly-home", + "LANG": "C.UTF-8", + "LC_ALL": "C.UTF-8", + "PATH": "/usr/local/bin:/usr/bin:/bin", + "XDG_CONFIG_HOME": "/tmp/fly-xdg", + } + boundary = FlyctlMachineBoundary(environment, APP, cwd=self.base) + command = ("flyctl", "apps", "list", "--json") + completed = subprocess.CompletedProcess(command, 0, "{}", "") + with mock.patch("subprocess.run", return_value=completed) as run: + boundary.run(command, timeout=7) + _, kwargs = run.call_args + self.assertIs(kwargs["shell"], False) + self.assertEqual(kwargs["env"], environment) + self.assertEqual(kwargs["timeout"], 7) + self.assertNotIn(token, " ".join(run.call_args.args[0])) + with self.assertRaisesRegex(FlyTransportError, "not allowed"): + boundary.run(("flyctl", "auth", "token"), timeout=7) + with self.assertRaisesRegex(FlyTransportError, "not allowed"): + boundary.run(("flyctl", "apps", "list", "--access-token", token), timeout=7) + for argument in ( + f"--access-token={token}", + f"-t{token}", + "--config=/tmp/foreign.toml", + ): + with ( + self.subTest(argument=argument), + self.assertRaisesRegex(FlyTransportError, "not allowed"), + ): + boundary.run(("flyctl", "apps", "list", "--json", argument), timeout=7) + with self.assertRaisesRegex(FlyTransportError, "not allowed"): + boundary.run(("flyctl", "apps", "destroy", "production-app", "--yes"), timeout=7) + with self.assertRaisesRegex(FlyTransportError, "not allowed"): + boundary.run( + ("flyctl", "machine", "list", "--app", "production-app", "--json"), + timeout=7, + ) + with self.assertRaisesRegex(FlyTransportError, "not allowed"): + boundary.run( + ("flyctl", "machine", "list", "-aproduction-app", "--json"), + timeout=7, + ) + with self.assertRaisesRegex(FlyTransportError, "not allowed"): + boundary.run(("flyctl", "machine", "list", "--json"), timeout=7) + + def test_concrete_api_is_scoped_to_owned_app_and_resource_shapes(self) -> None: + environment = { + "FLY_API_TOKEN": "fixture-token", + "HOME": "/tmp/fly-home", + "LANG": "C.UTF-8", + "LC_ALL": "C.UTF-8", + "PATH": "/usr/local/bin:/usr/bin:/bin", + "XDG_CONFIG_HOME": "/tmp/fly-xdg", + } + boundary = FlyctlMachineBoundary(environment, APP) + for path in ( + "/v1/apps/production-app", + f"/v1/apps/{APP}/machines/not-a-machine", + f"/v1/apps/{APP}/volumes/production-volume", + f"/v1/apps/{APP}/machines/{MACHINE_ID}/metadata", + ): + with self.subTest(path=path), self.assertRaisesRegex(FlyTransportError, "not allowed"): + boundary.api_json(path, timeout=1) + + def test_concrete_api_failure_does_not_retain_token_canary(self) -> None: + token = "FlyV1 secret-canary" + environment = { + "FLY_API_TOKEN": token, + "HOME": "/tmp/fly-home", + "LANG": "C.UTF-8", + "LC_ALL": "C.UTF-8", + "PATH": "/usr/local/bin:/usr/bin:/bin", + "XDG_CONFIG_HOME": "/tmp/fly-xdg", + } + + def fail(*_args: object, **_kwargs: object) -> object: + raise OSError(token) + + boundary = FlyctlMachineBoundary(environment, APP, urlopen=fail) + with self.assertRaises(FlyTransportError) as raised: + boundary.machine_state(APP, MACHINE_ID, timeout=1) + self.assertNotIn(token, str(raised.exception)) + self.assertIsNone(raised.exception.__cause__) + + def test_live_surfaces_remain_statically_unwired(self) -> None: + repository = ROOT.parent + operator = ( + repository / "benchmarks/harness/graphforge_bench/qualification_operator.py" + ).read_text(encoding="utf-8") + registry = (repository / "config/gate-registry.json").read_text(encoding="utf-8") + workflow = (repository / ".github/workflows/test.yml").read_text(encoding="utf-8") + self.assertNotIn("FlyProviderTransport", operator) + self.assertIn("progressive-ladder execution is unavailable", operator) + progressive = next( + gate + for gate in json.loads(registry)["operator_gates"] + if gate["id"] == "progressive-ladder" + ) + self.assertEqual(progressive["control_plane"], "pulumi_esc") + self.assertNotIn("execute_attempt", workflow) + + +if __name__ == "__main__": + unittest.main()