diff --git a/.kiro/steering/kube-ax.md b/.kiro/steering/kube-ax.md index 20d7447..b3d6997 100644 --- a/.kiro/steering/kube-ax.md +++ b/.kiro/steering/kube-ax.md @@ -18,8 +18,10 @@ Verified by the operator on 2026-10-05. These constraints apply across all workl quota or a replacement for the worker memory limit. - `ousast-engine-search-verify-*` tasks have no egress. The requested exception for `files.because-security.com:443` is **pending**, not operationally verified. They - receive one presigned archive containing `checkout/`, `products/`, and `spec.json`, - built/exported by an executor. Only `VERIFY_INPUT_URL` and `RESULT_URL` cross the + receive a presigned manifest for a gzip level-6 archive containing `checkout/`, + `products/`, and `spec.json`, built/exported by an executor. Compressed archives + retain the 1 GiB cap and travel in parts of at most 90 MB, each digest-checked + before assembly and extraction under `/workspace`. Only `VERIFY_INPUT_URL` and `RESULT_URL` cross the verifier boundary. Verifiers reject build recipes and do not install dependencies or fetch repositories. Executors finish and release their slots before verifier repetitions start. Both task types must be provisioned without model credentials. diff --git a/benchmarks/ax/batch.py b/benchmarks/ax/batch.py index ccfd8db..ba01e81 100644 --- a/benchmarks/ax/batch.py +++ b/benchmarks/ax/batch.py @@ -270,13 +270,35 @@ class AXLane: prefix = f"engine-queue/{self.run}/{key_id}/attempt-{attempt}/" keys = [prefix + item.object_names.get(key, key.lower() + PRESIGNED_ARCHIVE_SUFFIX) for key in item.inputs] keys.append(prefix + item.result_object) + part_keys = [] expires = timedelta(seconds=self.workload.deadline + 600) worker_ip = None submitted = False started = self.clock() phase = "" try: - for key, data in zip(keys, item.inputs.values(), strict=False): + for env, key, data in zip(item.inputs, keys, item.inputs.values(), strict=False): + if env == "VERIFY_INPUT_URL": + import io + + from openultrasast.search.verify_task import MAX_PART_BYTES, upload_parts + + count = (len(data) + MAX_PART_BYTES - 1) // MAX_PART_BYTES + destinations = [key + f".part-{index}" for index in range(count)] + part_keys.extend(destinations) + put_urls = [self.store.presign_put(part, expires) for part in destinations] + get_urls = [self.store.presign_get(part, expires) for part in destinations] + if any(public_endpoint(url) != self.host for url in put_urls + get_urls): + raise EgressError("archive part hostname must match store egress policy") + from openultrasast.search.executor import URLTransport + + data = upload_parts( + io.BytesIO(data), lambda url, block: URLTransport().put(url, block, self.workload.deadline), put_urls + ) + data = json.loads(data) + for part, url in zip(data["parts"], get_urls, strict=True): + part["url"] = url + data = json.dumps(data).encode() self.store._put(key, data) urls = {env: self.store.presign_get(key, expires) for env, key in zip(item.inputs, keys, strict=False)} urls["RESULT_URL"] = self.store.presign_put(keys[-1], expires) @@ -337,7 +359,7 @@ class AXLane: # A successful list already confirmed a disappeared task is gone. if phase != "Missing": errors.append("task: " + exception_reason(exc)) - for key in keys: + for key in [*keys, *part_keys]: try: self.store._delete(key) except Exception as exc: @@ -391,8 +413,17 @@ def manifest(name, workload, item, urls, atespace="default", *, validate_name=Tr raise ValueError("unexpected task environment") if set(item.extra_env) & set(urls): raise ValueError("overlapping task environment") - for url in urls.values(): - public_endpoint(url) + for key, value in urls.items(): + if key == "PREPARED_PART_URLS": + from openultrasast.search.verify_task import MAX_PARTS + + parts = json.loads(value) + if not isinstance(parts, list) or len(parts) != MAX_PARTS: + raise ValueError("invalid prepared part URLs") + for url in parts: + public_endpoint(url) + else: + public_endpoint(value) env = [{"name": key, "value": str(value)} for key, value in {**urls, **item.extra_env}.items()] if len(json.dumps(env).encode()) >= 32768: raise ValueError("AX Task env exceeds the 32 KB limit") @@ -660,7 +691,10 @@ class SearchExecutorTask: self.lane, self.archive, self.deadline = lane, checkout_archive, deadline token = uuid.uuid4().hex[:20] self.name = "ousast-engine-search-exec-" + token - self.keys = [f"engine-queue/search-exec/{token}/{name}" for name in ("repo.tar", "command.json", "result.json", "prepared.tar.gz")] + self.keys = [f"engine-queue/search-exec/{token}/{name}" for name in ("repo.tar", "command.json", "result.json", "prepared.json")] + from openultrasast.search.verify_task import MAX_PARTS + + self.keys.extend(self.keys[3] + f".part-{index}" for index in range(MAX_PARTS)) self.submitted = False self.slot = False @@ -684,10 +718,12 @@ class SearchExecutorTask: "EXECUTOR_RESULT_PUT_URL": store.presign_put(self.keys[2], expires), "PREPARED_PUT_URL": store.presign_put(self.keys[3], expires), } + part_urls = [store.presign_put(key, expires) for key in self.keys[4:]] command_put = store.presign_put(self.keys[1], expires) result_get = store.presign_get(self.keys[2], expires) - if any(public_endpoint(url) != self.lane.host for url in [*urls.values(), command_put, result_get]): + if any(public_endpoint(url) != self.lane.host for url in [*urls.values(), *part_urls, command_put, result_get]): raise EgressError("executor presign hostname must match store egress policy") + urls["PREPARED_PART_URLS"] = json.dumps(part_urls) work = Workload(self.lane.workload.image, frozenset(urls), self.deadline, None) item = Item( self.name, @@ -720,7 +756,7 @@ class SearchExecutorTask: raise def prepare(self, spec): - """Return the executor-built single verifier input, checking its digest.""" + """Check the export manifest and every fragment before handing off the archive.""" remaining = min(300, self.end - time.monotonic()) result = self.client.submit("prepare_verification", spec, timeout_seconds=remaining) if result.get("status") != "ok": @@ -732,7 +768,19 @@ class SearchExecutorTask: stored = bounded_read(lambda: self.lane.store._get(self.keys[3]), self.end - time.monotonic()) if stored is None or hashlib.sha256(stored[0]).hexdigest() != result.get("sha256"): raise ValueError("executor archive missing or digest mismatch") - return stored[0] + import io + + from openultrasast.search.verify_task import assemble_parts + + def get(part): + value = bounded_read(lambda: self.lane.store._get(self.keys[4 + part["index"]]), self.end - time.monotonic()) + if value is None: + raise ValueError("executor archive part missing") + return value[0] + + archive = io.BytesIO() + assemble_parts(stored[0], archive, get) + return archive.getvalue() def __exit__(self, exc_type, exc, traceback): errors = [] diff --git a/benchmarks/search/pilot_run.py b/benchmarks/search/pilot_run.py index fc16a9b..0b6d1cf 100644 --- a/benchmarks/search/pilot_run.py +++ b/benchmarks/search/pilot_run.py @@ -146,6 +146,8 @@ class Pilot: stdout=diagnostic(getattr(command, "stdout", ""), maximum=1500, private=private), ready_seconds=getattr(command, "ready_seconds", 0), ) + if (phase == "verify" or phase.startswith("verify/")) and entry.get("stderr"): + entry["reason"] = entry["stderr"][-300:] self.record["failures"].append(entry) @contextmanager diff --git a/src/openultrasast/search/demo.py b/src/openultrasast/search/demo.py index 8ee4bf1..012a05e 100644 --- a/src/openultrasast/search/demo.py +++ b/src/openultrasast/search/demo.py @@ -59,7 +59,19 @@ MAX_VALUE_BYTES = 16384 # an enumerated build goal, never arbitrary package-manager flags or scripts. BUILD_RECIPES = { "none": (), - "pip": ("/usr/bin/python3", "-I", "-m", "pip", "install", "--no-index", "--no-deps", "--target", "/scratch/packages", "-r"), + "pip": ( + "/usr/bin/python3", + "-I", + "-m", + "pip", + "install", + "--no-index", + "--ignore-installed", + "--no-warn-script-location", + "--target", + "/scratch/packages", + "-r", + ), "npm": ("/usr/bin/npm", "ci", "--offline", "--ignore-scripts", "--prefix"), "composer": ("/usr/bin/composer", "install", "--no-plugins", "--no-scripts", "--no-interaction", "--working-dir"), "maven": ("/usr/bin/mvn", "--offline", "-f"), @@ -241,7 +253,8 @@ def preparation_command(recipe: str, path: Path, checkout: Path, products: Path) "-m", "pip", "install", - "--no-deps", + "--ignore-installed", + "--no-warn-script-location", "--target", str(products / "packages"), *(["-r", str(target)] if requirements else [str(target)]), diff --git a/src/openultrasast/search/executor.py b/src/openultrasast/search/executor.py index 037535d..9a3ad71 100644 --- a/src/openultrasast/search/executor.py +++ b/src/openultrasast/search/executor.py @@ -107,6 +107,7 @@ class InProcessExecutor: self.last_result: dict[str, Any] = {} self.isolation_mode = "unprobed" self.prepared_put_url: str | None = None + self.prepared_part_urls: list[str] = [] def submit( self, name: str, args: dict[str, Any], *, timeout_seconds: float = 30, limits: dict[str, Any] | None = None @@ -392,7 +393,7 @@ class InProcessExecutor: from .demo import preparation_command, validate_demo from .verify import _tree - from .verify_task import ArchiveLimit, bundle_ignore, pack_checkout, validate_spec + from .verify_task import MAX_PART_BYTES, ArchiveLimit, bundle_ignore, pack_checkout, upload_parts, validate_spec if not self.task_boundary or not self.prepared_put_url: raise ValueError("preparation requires an executor task output") @@ -442,7 +443,16 @@ class InProcessExecutor: digest = hashlib.file_digest(archive, "sha256").hexdigest() archive.seek(0) self.phase = "result upload" - URLTransport().put(self.prepared_put_url, archive, self._remaining()) + if self.prepared_part_urls: + manifest = upload_parts( + archive, lambda url, block: URLTransport().put(url, block, self._remaining()), self.prepared_part_urls + ) + URLTransport().put(self.prepared_put_url, manifest, self._remaining()) + digest = hashlib.sha256(manifest).hexdigest() + elif size <= MAX_PART_BYTES: + URLTransport().put(self.prepared_put_url, archive, self._remaining()) + else: + raise ValueError("multipart archive URLs required") return {"status": "ok", "archive_bytes": size, "sha256": digest} def _tree_bytes(self) -> int: @@ -656,6 +666,7 @@ def main() -> None: result_url = args.result_put_url or os.environ.get("EXECUTOR_RESULT_PUT_URL") repo_url = os.environ.get("REPO_URL") prepared_url = os.environ.get("PREPARED_PUT_URL") + prepared_parts = json.loads(os.environ.get("PREPARED_PART_URLS", "[]")) # Clearing Python's environment cannot remove initial /proc/self/environ. # Refuse a misconfigured task before any repository input is opened. inherited = sorted( @@ -678,6 +689,7 @@ def main() -> None: unpack_checkout(download(repo_url, 128 * 1024**2, readiness_budget=min(60, args.deadline)), args.repo) executor = InProcessExecutor(args.repo, task_boundary=args.task_boundary) executor.prepared_put_url = prepared_url + executor.prepared_part_urls = prepared_parts serve(executor, command_url, result_url, deadline_seconds=args.deadline) diff --git a/src/openultrasast/search/verify.py b/src/openultrasast/search/verify.py index 5470819..685992d 100644 --- a/src/openultrasast/search/verify.py +++ b/src/openultrasast/search/verify.py @@ -18,6 +18,7 @@ import stat import subprocess import tempfile import time +import zipfile from collections.abc import Callable from concurrent.futures import ThreadPoolExecutor from contextlib import suppress @@ -380,6 +381,21 @@ def _side(side: Side, demo: Path, family: str, timeout: int, browser: BrowserExe ) command = ("/usr/bin/python3", "-I", "-c", bootstrap, "/workspace/" + start_spec["path"], *start_spec["arguments"]) + if start_spec["runtime"] == "java": + # -jar ignores CLASSPATH. Use the jar's declared main class + # with the verifier-owned product directories and jars. + with zipfile.ZipFile(entry) as jar, jar.open("META-INF/MANIFEST.MF") as jar_manifest: + metadata = jar_manifest.read(65537) + if len(metadata) > 65536: + return record("could_not_build", "oversized Java manifest") + metadata = metadata.replace(b"\r\n", b"\n").replace(b"\n ", b"") + main = re.search(rb"(?mi)^Main-Class: ([\w.$]+)\s*$", metadata.split(b"\n\n", 1)[0]) + if main is None: + return record("could_not_build", "Java jar missing Main-Class") + classpath = ["/workspace/" + start_spec["path"], "/build/", "/build/classes"] + classpath.extend("/build/" + str(path.relative_to(products)) for path in sorted(products.rglob("*.jar"))) + command = ("/usr/bin/java", "-cp", ":".join(classpath), main[1].decode(), *start_spec["arguments"]) + oracle = oracle_for(family, browser) assert oracle is not None try: @@ -393,6 +409,8 @@ def _side(side: Side, demo: Path, family: str, timeout: int, browser: BrowserExe }, **oracle.prepare(canary), } + if start_spec["runtime"] == "node": + app_env["NODE_PATH"] = "/build/node_modules" app_scratch = canary.root if family == "command" else root / "app" app_scratch.mkdir(exist_ok=True) fixture = canary.root @@ -483,7 +501,7 @@ def _side(side: Side, demo: Path, family: str, timeout: int, browser: BrowserExe return record("could_not_run", str(exc), exc.phase, exc) except task_storage.ScratchLimit as exc: return record("could_not_build", exception_reason(exc)) - except (OSError, ValueError, AssertionError, KeyError, IndexError, TypeError) as exc: + except (OSError, ValueError, AssertionError, KeyError, IndexError, TypeError, zipfile.BadZipFile) as exc: return record("could_not_run", "verification unavailable or refused: " + exception_reason(exc)) diff --git a/src/openultrasast/search/verify_task.py b/src/openultrasast/search/verify_task.py index 77e3544..d036921 100644 --- a/src/openultrasast/search/verify_task.py +++ b/src/openultrasast/search/verify_task.py @@ -3,6 +3,7 @@ from __future__ import annotations import gzip +import hashlib import io import ipaddress import json @@ -48,6 +49,9 @@ def public_endpoint(url: str) -> str: MAX_ARCHIVE_BYTES = 1024**3 +MAX_PART_BYTES = 90_000_000 +MAX_PARTS = (MAX_ARCHIVE_BYTES + MAX_PART_BYTES - 1) // MAX_PART_BYTES +MAX_MANIFEST_BYTES = 65536 MAX_CHECKOUT_BYTES = 2 * 1024**3 MAX_FILES = 20000 MAX_SPEC_BYTES = 70000 @@ -78,7 +82,7 @@ def pack_checkout(root: Path, output: BinaryIO | None = None) -> bytes: buffer = output if output is not None else io.BytesIO() total, count = 0, 0 directories: dict[str, int] = {} - with tarfile.open(fileobj=buffer, mode="w:gz") as archive: + with tarfile.open(fileobj=buffer, mode="w:gz", compresslevel=6) as archive: for parent, dirs, files in os.walk(root, followlinks=False): ignored = bundle_ignore(parent, dirs + files) dirs[:] = sorted(set(dirs) - ignored) @@ -101,13 +105,70 @@ def pack_checkout(root: Path, output: BinaryIO | None = None) -> bytes: return cast(io.BytesIO, buffer).getvalue() if output is None else b"" -def extract_checkout(data: bytes, destination: Path) -> None: - if len(data) > MAX_ARCHIVE_BYTES: +def upload_parts(source: BinaryIO, put: Any, urls: list[str]) -> bytes: + """Publish bounded gzip fragments, then return the ordered digest manifest.""" + source.seek(0, 2) + size = source.tell() + if not 0 < size <= MAX_ARCHIVE_BYTES: + raise ValueError("archive exceeds size limit") + if len(urls) < (size + MAX_PART_BYTES - 1) // MAX_PART_BYTES: + raise ValueError("missing archive part URLs") + source.seek(0) + parts: list[dict[str, Any]] = [] + while block := source.read(MAX_PART_BYTES): + index = len(parts) + put(urls[index], block) + parts.append({"index": index, "size": len(block), "sha256": hashlib.sha256(block).hexdigest()}) + return json.dumps({"format": "gzip-parts-v1", "size": size, "parts": parts}).encode() + + +def assemble_parts(data: bytes, output: BinaryIO, get: Any) -> None: + """Validate the complete manifest before fetching, and each part before use.""" + if len(data) > MAX_MANIFEST_BYTES: + raise ValueError("oversized archive manifest") + manifest = json.loads(data) + if not isinstance(manifest, dict) or manifest.get("format") != "gzip-parts-v1": + raise ValueError("invalid archive manifest") + parts = manifest.get("parts") + size = manifest.get("size") + if type(size) is not int or not 0 < size <= MAX_ARCHIVE_BYTES: + raise ValueError("archive exceeds size limit") + if not isinstance(parts, list) or not 1 <= len(parts) <= MAX_PARTS: + raise ValueError("invalid archive parts") + total = 0 + for index, part in enumerate(parts): + if ( + not isinstance(part, dict) + or type(part.get("index")) is not int + or part["index"] != index + or type(part.get("size")) is not int + or not 0 < part["size"] <= MAX_PART_BYTES + or not isinstance(part.get("sha256"), str) + or len(part["sha256"]) != 64 + ): + raise ValueError("invalid archive part") + total += part["size"] + if total != size: + raise ValueError("archive part sizes mismatch") + for part in parts: + block = get(part) + if len(block) != part["size"] or hashlib.sha256(block).hexdigest() != part["sha256"]: + raise ValueError("archive part digest mismatch") + output.write(block) + output.seek(0) + + +def extract_checkout(data: bytes | BinaryIO, destination: Path) -> None: + compressed_source = io.BytesIO(data) if isinstance(data, bytes) else data + compressed_source.seek(0, 2) + size = compressed_source.tell() + compressed_source.seek(0) + if size > MAX_ARCHIVE_BYTES: raise ValueError("archive exceeds size limit") # Bound expansion before tarfile parses PAX headers, using disk rather than # allocating the entire expanded tree in worker RAM. with tempfile.TemporaryFile() as expanded: - with gzip.GzipFile(fileobj=io.BytesIO(data)) as compressed: + with gzip.GzipFile(fileobj=compressed_source) as compressed: maximum = MAX_CHECKOUT_BYTES + MAX_FILES * 4096 while block := compressed.read(min(65536, maximum - expanded.tell() + 1)): expanded.write(block) @@ -232,7 +293,12 @@ def task_main() -> None: with tempfile.TemporaryDirectory(prefix="ousast-verify-task-") as temporary: root = Path(temporary) try: - extract_checkout(archive, root / "input") + if archive.startswith(b"\x1f\x8b"): + extract_checkout(archive, root / "input") + else: + with tempfile.TemporaryFile() as compressed: + assemble_parts(archive, compressed, lambda part: download(part["url"], part["size"])) + extract_checkout(compressed, root / "input") task_storage.check(task_storage.WORKSPACE if task_storage.WORKSPACE.is_dir() else root) bundle = root / "input" if {p.name for p in bundle.iterdir()} != {"checkout", "products", "spec.json"}: diff --git a/tests/test_ax_batch.py b/tests/test_ax_batch.py index 78f083f..32701ac 100644 --- a/tests/test_ax_batch.py +++ b/tests/test_ax_batch.py @@ -217,7 +217,13 @@ def test_search_executor_task_links_and_cleanup(monkeypatch, failure): assert session.client.command_url.endswith("?put") doc = calls[0][1]["manifest"] assert doc["spec"]["command"][2] == "openultrasast.search.executor" - assert {r["name"] for r in doc["spec"]["env"]} == {"COMMAND_GET_URL", "EXECUTOR_RESULT_PUT_URL", "REPO_URL", "PREPARED_PUT_URL"} + assert {r["name"] for r in doc["spec"]["env"]} == { + "COMMAND_GET_URL", + "EXECUTOR_RESULT_PUT_URL", + "REPO_URL", + "PREPARED_PUT_URL", + "PREPARED_PART_URLS", + } assert "API_KEY" not in json.dumps(doc) if failure: @@ -227,7 +233,9 @@ def test_search_executor_task_links_and_cleanup(monkeypatch, failure): monkeypatch.setattr(session.client, "submit", submit) session.client.submit("list_files", {}) assert calls[-1][0][:2] == ("delete", "task") - assert len(lane.store.deleted) == 4 + from openultrasast.search.verify_task import MAX_PARTS + + assert len(lane.store.deleted) == 4 + MAX_PARTS def test_capacity_backpressure_retries_same_manifest_until_deadline(tmp_path, monkeypatch): @@ -419,3 +427,88 @@ def test_search_preparation_failure_retains_command_record(): task.prepare({}) assert caught.value.phase == "build" and caught.value.exit_code == 8 assert "dependency missing" in caught.value.stderr and "hidden" not in caught.value.stderr + + +def test_verifier_lane_uploads_parts_and_task_reassembles(tmp_path, monkeypatch): + import io + import subprocess + from types import SimpleNamespace + + from benchmarks.ax.batch import AXLane + from openultrasast.search import executor, verify_task + + monkeypatch.setattr(verify_task, "MAX_PART_BYTES", 100) + monkeypatch.setenv("S3_ENDPOINT", "https://files.example") + source = tmp_path / "source" + source.mkdir() + (source / "input").write_bytes(bytes(range(256)) * 4) + blob = verify_task.pack_checkout(source) + objects, puts = {}, [] + + class Store: + def _put(self, key, data): + objects["https://files.example/" + key] = data + + def _get(self, key): + return b"completed", None + + def _delete(self, key): + objects.pop("https://files.example/" + key, None) + + def presign_get(self, key, expiry): + return "https://files.example/" + key + + presign_put = presign_get + + def put(self, url, block, timeout): + assert len(block) <= 100 + puts.append(url) + objects[url] = block + + monkeypatch.setattr(executor.URLTransport, "put", put) + work = Workload(IMAGE, frozenset({"VERIFY_INPUT_URL", "RESULT_URL"}), 10, lambda *args: None, kind="search-verify") + lane = AXLane(SimpleNamespace(ax_bin="fake", kubeconfig=None, atespace="default"), work, Store()) + + def ax(*args, **kwargs): + if args[0] == "apply": + env = {value["name"]: value["value"] for value in kwargs["manifest"]["spec"]["env"]} + assert set(env) == {"VERIFY_INPUT_URL", "RESULT_URL"} + restored = io.BytesIO() + verify_task.assemble_parts(objects[env["VERIFY_INPUT_URL"]], restored, lambda part: objects[part["url"]]) + assert restored.getvalue() == blob + return subprocess.CompletedProcess([], 0, "", "") + + monkeypatch.setattr(lane, "ax", ax) + request = Item("verify", {"VERIFY_INPUT_URL": blob}, "result.json", ("python3",), {}, "digest") + output, _ = lane._execute_attempt(request, tmp_path, 0, []) + assert output.returncode == 0, output + assert len(puts) > 1 + assert not objects # fragment cleanup is part of the task lifecycle + + +@pytest.mark.parametrize("corrupt", [False, True]) +def test_executor_prepare_checks_part_digest(monkeypatch, corrupt): + import hashlib + import io + import time + from types import SimpleNamespace + + from benchmarks.ax.batch import SearchExecutorTask + from openultrasast.search import verify_task + + monkeypatch.setattr(verify_task, "MAX_PART_BYTES", 100) + blob = b"compressed bytes" * 30 + objects = {} + lane = SimpleNamespace(store=SimpleNamespace(_get=lambda key: (objects[key], None))) + task = SearchExecutorTask(lane, b"checkout") + metadata = verify_task.upload_parts(io.BytesIO(blob), objects.__setitem__, task.keys[4:]) + objects[task.keys[3]] = metadata + if corrupt: + objects[task.keys[4]] = b"corrupt" + task.end = time.monotonic() + 60 + task.client = SimpleNamespace(submit=lambda *args, **kwargs: {"status": "ok", "sha256": hashlib.sha256(metadata).hexdigest()}) + if corrupt: + with pytest.raises(ValueError, match="part digest mismatch"): + task.prepare({}) + else: + assert task.prepare({}) == blob diff --git a/tests/test_search_executor_smoke.py b/tests/test_search_executor_smoke.py index ca93bc8..04e3fc8 100644 --- a/tests/test_search_executor_smoke.py +++ b/tests/test_search_executor_smoke.py @@ -77,7 +77,13 @@ def setup(tmp_path, monkeypatch): if args[0] == "apply": doc = kwargs["manifest"] assert doc["metadata"]["name"].startswith("ousast-engine-search-exec-") - assert {x["name"] for x in doc["spec"]["env"]} == {"REPO_URL", "COMMAND_GET_URL", "EXECUTOR_RESULT_PUT_URL", "PREPARED_PUT_URL"} + assert {x["name"] for x in doc["spec"]["env"]} == { + "REPO_URL", + "COMMAND_GET_URL", + "EXECUTOR_RESULT_PUT_URL", + "PREPARED_PUT_URL", + "PREPARED_PART_URLS", + } lane.ax = ax return executor, lane, events diff --git a/tests/test_search_pilot_run.py b/tests/test_search_pilot_run.py index f4523df..5bdc9f5 100644 --- a/tests/test_search_pilot_run.py +++ b/tests/test_search_pilot_run.py @@ -536,7 +536,9 @@ def test_pilot_executor_deletes_task_when_submit_times_out(tmp_path, monkeypatch with pytest.raises(TimeoutError, match="upload deadline"), pilot.executor("a" * 40, deadline=60) as session: session.client.submit("list_files", {}) assert calls[-1][0][:2] == ("delete", "task") - assert len(deleted) == 4 + from openultrasast.search.verify_task import MAX_PARTS + + assert len(deleted) == 4 + MAX_PARTS assert pilot.record["executor_task_metrics"][-1]["end"] == "execution_failure" @@ -593,3 +595,15 @@ def test_ax_pilot_uses_verification_task_limits_for_deadline(tmp_path, monkeypat assert captured == [17] assert deadlines == [17 * (len(schema["steps"]) + 1) + 65] assert pilot.lane.workload is None + + +def test_verify_failure_reason_is_last_300_stderr_characters(tmp_path): + from openultrasast.search.verify import SideRecord + + pilot = make_pilot(tmp_path) + stderr = "Traceback:\n" + "middle of traceback " * 200 + "\nModuleNotFoundError: No module named 'sniffio'" + side = SideRecord("could_not_run", (), 0, "app readiness failed", phase="start", exit_code=1, stderr=stderr) + pilot.failure("verify/could_not_run", side.reason, side) + failure = pilot.record["failures"][-1] + assert failure["reason"] == stderr[-300:] + assert failure["reason"].endswith("ModuleNotFoundError: No module named 'sniffio'") diff --git a/tests/test_search_task_storage.py b/tests/test_search_task_storage.py index da60d91..315fbc3 100644 --- a/tests/test_search_task_storage.py +++ b/tests/test_search_task_storage.py @@ -335,13 +335,23 @@ def test_verifier_app_virtual_memory(workspace): @pytest.mark.parametrize( "recipe,manifest,extra,expected", [ - ("pip", "pyproject.toml", {}, ["-I", "-m", "pip", "install", "--no-deps", "--target", "PACKAGES", "PROJECT"]), - ("pip", "requirements.txt", {}, ["-I", "-m", "pip", "install", "--no-deps", "--target", "PACKAGES", "-r", "MANIFEST"]), + ( + "pip", + "pyproject.toml", + {}, + ["-I", "-m", "pip", "install", "--ignore-installed", "--no-warn-script-location", "--target", "PACKAGES", "PROJECT"], + ), + ( + "pip", + "requirements.txt", + {}, + ["-I", "-m", "pip", "install", "--ignore-installed", "--no-warn-script-location", "--target", "PACKAGES", "-r", "MANIFEST"], + ), ( "pip", "pyproject.toml", {"requirements.txt": ""}, - ["-I", "-m", "pip", "install", "--no-deps", "--target", "PACKAGES", "-r", "REQUIREMENTS"], + ["-I", "-m", "pip", "install", "--ignore-installed", "--no-warn-script-location", "--target", "PACKAGES", "-r", "REQUIREMENTS"], ), ("npm", "package.json", {}, ["install", "--ignore-scripts"]), ("npm", "package.json", {"package-lock.json": "{}"}, ["ci", "--ignore-scripts"]), @@ -481,27 +491,30 @@ def test_large_export_spools_to_workspace_and_excludes_caches(workspace, monkeyp path = repo / "nested" / directory path.mkdir(parents=True) (path / "excluded").write_bytes(b"cache") - seen = {} + uploaded = {} def upload(self, url, data, timeout): - assert not isinstance(data, bytes) - assert str(workspace) in os.readlink(f"/proc/self/fd/{data.fileno()}") - seen["size"] = os.fstat(data.fileno()).st_size - seen["sha256"] = hashlib.file_digest(data, "sha256").hexdigest() - data.seek(0) - with tarfile.open(fileobj=data, mode="r:gz") as archive: - names = archive.getnames() - assert "checkout/large.bin" in names - assert not any(name.endswith("excluded") for name in names) - data.seek(0) + assert isinstance(data, bytes) + assert len(data) <= verify_task.MAX_PART_BYTES + uploaded[url] = data monkeypatch.setattr(executor.URLTransport, "put", upload) app = executor.InProcessExecutor(repo, task_boundary=True) app.prepared_put_url = "https://files.example/output" + app.prepared_part_urls = [f"https://files.example/part-{i}" for i in range(verify_task.MAX_PARTS)] result = app.submit("prepare_verification", spec(), timeout_seconds=60) assert result["status"] == "ok", result - assert 128 * 1024**2 < result["archive_bytes"] == seen["size"] - assert result["sha256"] == seen["sha256"] + manifest = uploaded[app.prepared_put_url] + assert result["sha256"] == hashlib.sha256(manifest).hexdigest() + assert len(uploaded) == 3 # two bounded parts and the manifest + assert json.loads(manifest)["size"] == result["archive_bytes"] > 128 * 1024**2 + with verify_task.tempfile.TemporaryFile(dir=workspace) as archive: + assert str(workspace) in os.readlink(f"/proc/self/fd/{archive.fileno()}") + verify_task.assemble_parts(manifest, archive, lambda part: uploaded[app.prepared_part_urls[part["index"]]]) + with tarfile.open(fileobj=archive, mode="r:gz") as contents: + names = contents.getnames() + assert "checkout/large.bin" in names + assert not any(name.endswith("excluded") for name in names) assert verify_task.MAX_ARCHIVE_BYTES == 1024**3 @@ -525,3 +538,61 @@ def test_export_over_cap_reports_size_and_five_relative_directories(workspace, m for i in range(2, 6): assert f"checkout/dir{i} (" in result["reason"] assert "dir0" not in result["reason"] and "dir1" not in result["reason"] + + +@pytest.mark.parametrize("location", ["products", "checkout"]) +def test_node_resolves_bundled_module(workspace, location): + if not Path("/usr/bin/node").exists(): + pytest.skip("node unavailable") + repo, products = workspace / "checkout", workspace / "products" + (repo / "bin").mkdir(parents=True) + products.mkdir() + module = (products if location == "products" else repo) / "node_modules/bash-color" + module.mkdir(parents=True) + (module / "index.js").write_text("module.exports = 'bundled dependency';") + if location == "checkout": + destination = workspace / "bundle-check" + verify_task.extract_checkout(verify_task.pack_checkout(repo), destination) + assert (destination / "node_modules/bash-color/index.js").read_text() == (module / "index.js").read_text() + (repo / "bin/glance.js").write_text("if(require('bash-color') !== 'bundled dependency') process.exit(7); console.log('ready');") + value = spec()["demo"] + value["start"].update(runtime="node", path="bin/glance.js") + demo = workspace / "demo.json" + demo.write_text(json.dumps(value)) + result = verify_side_task(Side(repo, products=products), demo, "path") + assert result.outcome == "observed", result + assert result.exit_code is None or result.exit_code == 0 + + +def test_java_classpath_includes_products(workspace, monkeypatch): + import importlib + import zipfile + + from openultrasast.sandbox import SandboxResult + + verifier = importlib.import_module("openultrasast.search.verify") + repo, products = workspace / "checkout", workspace / "products" + repo.mkdir() + (products / "lib").mkdir(parents=True) + (products / "lib/dependency.jar").write_bytes(b"dependency") + with zipfile.ZipFile(repo / "app.jar", "w") as jar: + jar.writestr("META-INF/MANIFEST.MF", "Manifest-Version: 1.0\nMain-Class: example.\n Main\n\n") + value = spec()["demo"] + value["start"].update(runtime="java", path="app.jar") + demo = workspace / "demo.json" + demo.write_text(json.dumps(value)) + calls = [] + + def run(job, **kwargs): + calls.append(job.command) + assert job.command[:2] == ("/usr/bin/java", "-cp") + assert "/build/lib/dependency.jar" in job.command[2] + assert "/build/classes" in job.command[2] + assert job.command[3] == "example.Main" + assert (kwargs["mounts"]["/build"] / "lib/dependency.jar").read_bytes() == b"dependency" + return SandboxResult(0, "ready", "", False) + + monkeypatch.setattr(verifier, "_run_task", run) + result = verify_side_task(Side(repo, products=products), demo, "path") + assert result.outcome == "observed", result + assert len(calls) == 2 diff --git a/tests/test_search_verify.py b/tests/test_search_verify.py index 46b02c1..9653cfe 100644 --- a/tests/test_search_verify.py +++ b/tests/test_search_verify.py @@ -325,7 +325,7 @@ def test_pip_products_imported_by_runtime_and_immutable(pair, monkeypatch): def runner(job, **kwargs): if "pip" in job.command: - assert "--no-index" in job.command and "--no-deps" in job.command + assert "--no-index" in job.command and "--ignore-installed" in job.command and "--no-warn-script-location" in job.command packages = kwargs["scratch"] / "packages" packages.mkdir() (packages / "dependency.py").write_text("VALUE = 42\n") diff --git a/tests/test_search_verify_transport.py b/tests/test_search_verify_transport.py index 924f8ef..ba72c64 100644 --- a/tests/test_search_verify_transport.py +++ b/tests/test_search_verify_transport.py @@ -67,7 +67,8 @@ def test_extract_rejects_escape_and_links(tmp_path, name, kind): entry.extract_checkout(data.getvalue(), tmp_path / "repo") -def test_entry_reads_input_and_scrubs_environment(tmp_path, monkeypatch, capsys): +@pytest.mark.parametrize("multipart", [False, True]) +def test_entry_reads_input_and_scrubs_environment(tmp_path, monkeypatch, capsys, multipart): root = tmp_path / "repo" root.mkdir() (root / "app.py").write_text("print(1)") @@ -80,9 +81,18 @@ def test_entry_reads_input_and_scrubs_environment(tmp_path, monkeypatch, capsys) (bundle / "spec.json").write_text(json.dumps({"demo": demo(), "family": "injection", "timeout_seconds": 2})) blob = entry.pack_checkout(bundle) + objects = {} + if multipart: + urls = ["https://store.example/part"] + manifest = json.loads(entry.upload_parts(io.BytesIO(blob), objects.__setitem__, urls)) + manifest["parts"][0]["url"] = urls[0] + objects["https://store.example/input"] = json.dumps(manifest).encode() + else: + objects["https://store.example/input"] = blob + def download(url, limit): - assert limit == 1024**3 - return blob + assert len(objects[url]) <= limit + return objects[url] monkeypatch.setattr(entry, "download", download) outputs = [] @@ -265,3 +275,41 @@ def test_verifier_deadline_uses_mode_budget_and_all_steps(tmp_path, monkeypatch, ) dispatch(side=Side(tmp_path), demo=schema, family="injection", timeout_seconds=timeout, fresh_task=True) assert workloads[0].deadline == verification_run_seconds(schema, timeout) + 60 == timeout * 9 + 65 + + +@pytest.mark.parametrize("corrupt", [False, True]) +def test_multipart_digest_checked_before_extraction(tmp_path, monkeypatch, corrupt): + import hashlib + + monkeypatch.setattr(entry, "MAX_PART_BYTES", 100) + source = tmp_path / "source" + source.mkdir() + (source / "input").write_bytes(bytes(range(256)) * 4) + blob = entry.pack_checkout(source) + objects = {} + urls = [f"https://store.example/part-{index}" for index in range(entry.MAX_PARTS)] + manifest = entry.upload_parts(io.BytesIO(blob), objects.__setitem__, urls) + parts = json.loads(manifest)["parts"] + assert len(objects) == len(parts) > 1 + assert all(len(value) <= 100 for value in objects.values()) + assert all(hashlib.sha256(objects[urls[p["index"]]]).hexdigest() == p["sha256"] for p in parts) + if corrupt: + objects[urls[1]] = b"x" * len(objects[urls[1]]) + assembled = io.BytesIO() + if corrupt: + with pytest.raises(ValueError, match="part digest mismatch"): + entry.assemble_parts(manifest, assembled, lambda part: objects[urls[part["index"]]]) + assert not (tmp_path / "restored").exists() + else: + entry.assemble_parts(manifest, assembled, lambda part: objects[urls[part["index"]]]) + entry.extract_checkout(assembled, tmp_path / "restored") + assert (tmp_path / "restored/input").read_bytes() == (source / "input").read_bytes() + + +def test_multipart_rejects_total_cap_before_fetch(monkeypatch): + monkeypatch.setattr(entry, "MAX_ARCHIVE_BYTES", 10) + with pytest.raises(ValueError, match="size limit"): + entry.upload_parts(io.BytesIO(b"x" * 11), lambda *args: pytest.fail("must not upload"), ["unused"]) + manifest = json.dumps({"format": "gzip-parts-v1", "size": 11, "parts": []}).encode() + with pytest.raises(ValueError, match="size limit"): + entry.assemble_parts(manifest, io.BytesIO(), lambda *args: pytest.fail("must not fetch"))