diff --git a/benchmarks/ax/batch.py b/benchmarks/ax/batch.py index 9f387d9..81e3233 100644 --- a/benchmarks/ax/batch.py +++ b/benchmarks/ax/batch.py @@ -25,6 +25,7 @@ import yaml from openultrasast.config import load_dotenv from openultrasast.plane.memory import PRESIGNED_ARCHIVE_SUFFIX, S3Store, open_store +from openultrasast.search.task_storage import diagnostic, exception_reason class EgressError(ValueError): @@ -54,6 +55,12 @@ def apply_when_available(lane, document, end): lane.ax("apply", "-f", "-", manifest=document, timeout=min(30, end - lane.clock())) return except NoFreeWorkers: + # A failed apply is not proof that the server did not create a task. + # Never send another create while the preceding task can still run. + try: + lane.ax("delete", "task", document["metadata"]["name"]) + except Exception as exc: + raise RuntimeError("capacity cleanup failed: " + exception_reason(exc)) from None lane.pause(min(random.uniform(15, 30), max(0, end - lane.clock()))) raise CapacityDeadline("no free workers available before deadline") @@ -185,7 +192,11 @@ class AXLane: if slot: AX_SLOTS.release() if done is None or done.returncode: - return {"status": "instrument_failure", "ax_attempts": attempts} + return { + "status": "instrument_failure", + "reason": diagnostic(done.stderr) if done is not None else "TimeoutError: task result deadline", + "ax_attempts": attempts, + } record = json.loads((output / "result.json").read_text()) return {**record, "worker_ip": worker, "ax_attempts": attempts} @@ -203,9 +214,11 @@ class AXLane: ) if done.returncode: if args[0] == "apply" and "no free workers available" in (done.stdout + done.stderr).lower(): - raise NoFreeWorkers - # AX errors may echo manifest URLs; report only verb and exit code. - raise ValueError(f"ax {args[0]} failed (exit {done.returncode})") + raise NoFreeWorkers("no free workers available") + if args[0] == "delete" and "notfound" in (done.stdout + done.stderr).lower().replace(" ", ""): + return done + # AX errors may echo manifest URLs; sanitize before recording. + raise ValueError(f"ax {args[0]} failed (exit {done.returncode}): " + diagnostic(done.stderr)) return done def _execute_attempt(self, item, output, attempt, attempts): @@ -266,13 +279,13 @@ class AXLane: raise SandboxFailure self.pause(min(2, max(0, end - self.clock()))) return None, worker_ip - except (subprocess.TimeoutExpired, TimeoutError): - return None, worker_ip + except (subprocess.TimeoutExpired, TimeoutError) as exc: + return subprocess.CompletedProcess([], 1, "", exception_reason(exc)), worker_ip except (SandboxFailure, CapacityDeadline): raise except Exception as exc: # Never expose exception strings which may contain object-scoped bearer URLs. - detail = str(exc) if isinstance(exc, EgressError) else type(exc).__name__ + detail = exception_reason(exc) return subprocess.CompletedProcess([], 1, "", "AX transport failed: " + detail), worker_ip finally: attempts.append({"name": name, "phase": phase, "seconds": self.clock() - started}) @@ -283,12 +296,12 @@ class AXLane: except Exception as exc: # A successful list already confirmed a disappeared task is gone. if phase != "Missing": - errors.append("task: " + type(exc).__name__) + errors.append("task: " + exception_reason(exc)) for key in keys: try: self.store._delete(key) except Exception as exc: - errors.append("object: " + type(exc).__name__) + errors.append("object: " + exception_reason(exc)) if errors: raise RuntimeError("AX cleanup failed; stop dispatching: " + ", ".join(errors)) @@ -414,6 +427,9 @@ def dispatch(items, workload, out, executors, *, lanes=("ax", "ax", "docker"), o if not isinstance(max_attempts, int) or not 1 <= max_attempts <= 2: raise ValueError("batch attempts must be one or two") out.mkdir(parents=True, exist_ok=True) + items = list(items) + if len({item.id for item in items}) != len(items): + raise ValueError("duplicate dispatch item id") pending, fallbacks = [], [] def path(item): @@ -465,15 +481,7 @@ def dispatch(items, workload, out, executors, *, lanes=("ax", "ax", "docker"), o def docker_diagnostic(text): """Bound diagnostics and remove bearer URLs, environment secrets and assignments.""" - text = str(text or "") - for key, value in os.environ.items(): - if len(value) >= 4 and any(word in key.upper() for word in ("TOKEN", "SECRET", "PASSWORD", "CREDENTIAL", "KEY")): - text = text.replace(value, "[redacted]") - text = re.sub(r"https?://[^\s\"']+", "[redacted-url]", text) - text = re.sub( - r"(?i)(authorization|bearer|password|token|secret|signature|credential|[\w-]*key)\s*[:= ]\s*[^\s,;]+", r"\1=[redacted]", text - ) - return "\n".join(text.splitlines()[-10:])[-4000:] + return diagnostic(text or "", maximum=4000) class DockerLane(AXLane): @@ -510,12 +518,12 @@ class DockerLane(AXLane): done = self.runner(command, capture_output=True, text=True, timeout=timeout) except (OSError, subprocess.SubprocessError) as exc: self._diagnostics.append( - {"step": command[1], "exit_code": None, "stderr": docker_diagnostic(getattr(exc, "stderr", "") or type(exc).__name__)} + {"step": command[1], "exit_code": None, "stderr": docker_diagnostic(getattr(exc, "stderr", "") or exception_reason(exc))} ) raise if done.returncode: self._diagnostics.append({"step": command[1], "exit_code": done.returncode, "stderr": docker_diagnostic(done.stderr)}) - raise ValueError(f"docker {command[1]} failed (exit {done.returncode})") + raise ValueError(f"docker {command[1]} failed (exit {done.returncode}): " + diagnostic(done.stderr)) return done def _prepare_image(self): @@ -625,6 +633,8 @@ class SearchExecutorTask: expires = timedelta(seconds=self.deadline + 600) store = self.lane.store try: + if isinstance(self.lane, DockerLane): + self.lane._run_docker(["docker", "image", "inspect", self.lane.workload.image], min(30, self.deadline)) store._put(self.keys[0], self.archive) urls = { "REPO_URL": store.presign_get(self.keys[0], expires), @@ -684,12 +694,12 @@ class SearchExecutorTask: try: self.lane.ax("delete", "task", self.name) except Exception as error: - errors.append(type(error).__name__) + errors.append(exception_reason(error)) for key in self.keys: try: self.lane.store._delete(key) except Exception as error: - errors.append(type(error).__name__) + errors.append(exception_reason(error)) if self.slot and not errors: AX_SLOTS.release() self.slot = False diff --git a/benchmarks/search/executor_smoke.py b/benchmarks/search/executor_smoke.py index 0a886ff..560e38f 100644 --- a/benchmarks/search/executor_smoke.py +++ b/benchmarks/search/executor_smoke.py @@ -1,7 +1,7 @@ """One no-model executor task; run as python -m benchmarks.search.executor_smoke. Runtime acquisition/install needs network inside the executor. Unit tests are offline. -Records contain measurements, never command output or presigned bearer URLs. +Records contain measurements and sanitized failure tails, never presigned bearer URLs. """ from __future__ import annotations @@ -16,6 +16,7 @@ from pathlib import Path from urllib.parse import urlsplit from benchmarks.ax.batch import AXLane, DockerLane, SearchExecutorTask, Workload, validate_image +from openultrasast.search.task_storage import diagnostic, exception_reason # URL and pin are positional shell arguments, never interpolated into shell code. # Retry the initial acquisition for delayed task egress activation (at most 60 s). @@ -55,6 +56,8 @@ def run_smoke(lane, repo_url, commit, *, deadline=900): "scratch_measurement": "sampled workspace allocated bytes plus command logs", "model_calls": 0, } + parts = urlsplit(repo_url) + private = tuple(p for p in parts.path.split("/") if p) + (parts.path.rsplit("/", 1)[-1].removesuffix(".git"),) started = time.monotonic() try: parts = urlsplit(repo_url) @@ -77,9 +80,18 @@ def run_smoke(lane, repo_url, commit, *, deadline=900): if name == "run": args = {**args, "timeout_seconds": timeout} before = time.monotonic() - result = task.client.submit( - name, args, timeout_seconds=timeout, limits={"memory_bytes": 2 * 1024**3, "disk_bytes": 2 * 1024**3} - ) + try: + result = task.client.submit( + name, args, timeout_seconds=timeout, limits={"memory_bytes": 2 * 1024**3, "disk_bytes": 2 * 1024**3} + ) + except Exception as exc: + result = { + "error": type(exc).__name__, + "reason": exception_reason(exc, private=private), + "phase": "timeout" if isinstance(exc, TimeoutError) else "result upload", + "exit_code": None, + "stderr": "", + } output_bytes = sum(result.get(k + "_bytes", len(result.get(k, "").encode())) for k in ("stdout", "stderr")) if name != "run": output_bytes = len(json.dumps(result).encode()) @@ -95,12 +107,27 @@ def run_smoke(lane, repo_url, commit, *, deadline=900): } if "input_bytes" in result: row["input_bytes"] = result["input_bytes"] + failed = ( + result.get("error") + or result.get("timed_out") + or result.get("status") == "could_not_build" + or (name == "run" and result.get("exit_code") != 0) + ) + if failed: + row.update( + phase=result.get("phase", "timeout" if result.get("timed_out") else "run"), + reason=diagnostic( + result.get("reason") or result.get("error") or f"CommandFailure: exit code {result.get('exit_code')}", + private=private, + ), + stderr=diagnostic(result.get("stderr", ""), maximum=4000, private=private), + ) record["commands"].append(row) record["scratch_peak_bytes"] = max(record["scratch_peak_bytes"], result.get("scratch_peak_bytes", 0)) if result.get("error") or result.get("timed_out") or result.get("status") == "could_not_build": - raise ValueError(step + " executor failure") + raise ValueError(step + " executor failure: " + row["reason"]) if name == "run" and result.get("exit_code") != 0: - raise ValueError(step + " nonzero or missing exit code") + raise ValueError(step + " nonzero or missing exit code: " + row["reason"]) if step == "install" and output_bytes == 0 and row["wall_seconds"] < 1: raise ValueError("empty fast install: exit 0 with empty output in under 1 s") return result @@ -130,28 +157,7 @@ def run_smoke(lane, repo_url, commit, *, deadline=900): record["status"] = "ok" except Exception as exc: record["status"] = "instrument_failure" - # Only our fixed diagnostic strings are safe; transport errors may contain URLs. - record["reason"] = ( - str(exc) - if isinstance(exc, ValueError) - and str(exc) - in { - "credential-free HTTPS repository URL required", - "full lowercase commit SHA required", - "no recognised install manifest", - "clone yielded zero files", - "file read lacks input size", - "trivial command did not print 1", - "stop not acknowledged", - "empty fast install: exit 0 with empty output in under 1 s", - *( - step + suffix - for step in ("clone", "list_files", "read_file", "grep", "install", "trivial", "stop") - for suffix in (" executor failure", " nonzero or missing exit code") - ), - } - else type(exc).__name__ - ) + record["reason"] = exception_reason(exc, private=private) record["wall_seconds"] = time.monotonic() - started return record @@ -173,7 +179,7 @@ def main(argv=None): lane = (DockerLane if args.lane == "docker" else AXLane)(args, Workload(args.image, frozenset(), args.deadline, None)) record = run_smoke(lane, args.repo_url, args.commit, deadline=args.deadline) except Exception as exc: - record = {"status": "instrument_failure", "reason": "lane setup: " + type(exc).__name__} + record = {"status": "instrument_failure", "reason": "lane setup: " + exception_reason(exc)} args.out.parent.mkdir(parents=True, exist_ok=True) args.out.write_text(json.dumps(record, indent=2, sort_keys=True) + "\n") print(json.dumps(record)) diff --git a/benchmarks/search/pilot_select.py b/benchmarks/search/pilot_select.py index 60c65f6..aff4321 100644 --- a/benchmarks/search/pilot_select.py +++ b/benchmarks/search/pilot_select.py @@ -19,6 +19,7 @@ from urllib.parse import urlsplit from benchmarks.unseen import eligibility, ledger from openultrasast.model.taxonomy import load_families +from openultrasast.search.task_storage import exception_reason LANGUAGES = ("python", "javascript", "typescript", "php", "java") FAMILIES = ("injection", "path", "output_encoding", "untrusted_destination") @@ -352,7 +353,16 @@ def main(argv=None) -> int: except Exception as exc: if created: private.unlink(missing_ok=True) - print(json.dumps({"status": "instrument_failure", "reason": type(exc).__name__})) + print( + json.dumps( + { + "status": "instrument_failure", + "reason": exception_reason( + exc, private=tuple(part for row in locals().get("rows", []) for part in str(row.get("repo", "")).split("/") if part) + ), + } + ) + ) return 1 diff --git a/plane/Dockerfile.search-task b/plane/Dockerfile.search-task index ebf9cff..947b271 100644 --- a/plane/Dockerfile.search-task +++ b/plane/Dockerfile.search-task @@ -28,5 +28,8 @@ RUN python3 -m venv /venv \ RUN groupadd --gid 10001 search \ && useradd --create-home --uid 10001 --gid 10001 search # gVisor is the outer boundary; retain inner bwrap when its capability probe succeeds. +# The Python base image sets GPG_KEY to the public CPython release-signing fingerprint (not a secret). +# Clear it so the executor's guard against inherited credentials stays strict and still passes. +ENV GPG_KEY= USER root # Do not replace the AX runner's ENTRYPOINT or CMD. diff --git a/src/openultrasast/search/executor.py b/src/openultrasast/search/executor.py index bd26624..0bf5e10 100644 --- a/src/openultrasast/search/executor.py +++ b/src/openultrasast/search/executor.py @@ -34,7 +34,7 @@ from . import _sandbox, task_storage MAX_COMMAND_BYTES = 65536 MAX_RESULT_BYTES = 65536 -SAFE_ENV = {"PATH": "/usr/bin:/bin", "LANG": "C.UTF-8"} +SAFE_ENV = {"PATH": "/venv/bin:/opt/java/openjdk/bin:/usr/local/bin:/usr/bin:/bin", "LANG": "C.UTF-8", "JAVA_HOME": "/opt/java/openjdk"} def clean_environment(_source: Mapping[str, str] | None = None) -> dict[str, str]: @@ -111,7 +111,7 @@ class InProcessExecutor: ) -> dict[str, Any]: result = self.execute(Command(self.seq + 1, name, args, timeout_seconds, limits or {})) if result.get("error") in {"ValueError", "KeyError", "TypeError", "FileNotFoundError", "OSError"}: - raise ValueError("repository command rejected") + raise ValueError(result.get("reason", "repository command rejected")) return result def execute(self, command: Command) -> dict[str, Any]: @@ -126,13 +126,28 @@ class InProcessExecutor: raise ValueError("out of order sequence") started = time.monotonic() self.deadline = started + command.timeout_seconds + self.phase = "run" try: result = self._execute(command) - except task_storage.ScratchLimit: + except task_storage.ScratchLimit as exc: self.stopped = True - result = {"status": "could_not_build", "reason": "scratch limit", "stopped": True} - except (ValueError, OSError, KeyError, TypeError, TimeoutError) as exc: - result = {"error": type(exc).__name__, "timed_out": isinstance(exc, TimeoutError)} + result = { + "status": "could_not_build", + "reason": task_storage.exception_reason(exc, private=(str(self.repo), self.repo.name)), + "phase": "scratch guard", + "exit_code": None, + "stderr": "", + "stopped": True, + } + except (ValueError, OSError, KeyError, TypeError, TimeoutError, subprocess.SubprocessError) as exc: + result = { + "error": type(exc).__name__, + "reason": task_storage.exception_reason(exc, private=(str(self.repo), self.repo.name)), + "phase": self.phase, + "exit_code": None, + "stderr": "", + "timed_out": isinstance(exc, TimeoutError), + } result["wall_seconds"] = time.monotonic() - started result = _bounded({"seq": command.seq, **result}, min(MAX_RESULT_BYTES, command.limits.get("result_bytes", MAX_RESULT_BYTES))) self.seq, self.last_digest, self.last_result = command.seq, digest, copy.deepcopy(result) @@ -179,7 +194,9 @@ class InProcessExecutor: def _execute(self, command: Command) -> dict[str, Any]: name, args = command.name, command.args if self.task_boundary: + self.phase = "scratch guard" task_storage.check(task_storage.WORKSPACE if task_storage.WORKSPACE.is_dir() else self.repo) + self.phase = "run" if name == "prepare_verification": return self._prepare_verification(args) if name == "stop": @@ -240,10 +257,18 @@ class InProcessExecutor: wall_timeout_seconds=self._remaining(), ) return { + **( + { + "phase": "timeout" if result.timed_out else "run", + "reason": "TimeoutError: command deadline" if result.timed_out else f"CommandFailure: exit code {result.exit_code}", + } + if result.timed_out or result.exit_code + else {} + ), "isolation_mode": self.isolation_mode, "exit_code": result.exit_code, "stdout": result.stdout[:16384], - "stderr": result.stderr[-16384:], + "stderr": task_storage.diagnostic(result.stderr, maximum=4000), "timed_out": result.timed_out, "truncated": len(result.stdout) > 16384 or len(result.stderr) > 16384, } @@ -259,12 +284,15 @@ class InProcessExecutor: initial_bytes = self._tree_bytes() guard_root = task_storage.WORKSPACE if task_storage.WORKSPACE.is_dir() else self.repo peak_bytes = task_storage.check(guard_root) + self.phase = "spawn" with tempfile.TemporaryFile() as out, tempfile.TemporaryFile() as err: proc = subprocess.Popen( argv, cwd=self.repo, env=clean_environment(), stdout=out, stderr=err, start_new_session=True, preexec_fn=constrain ) + self.phase = "run" timed_out = False disk_exceeded = False + failure = None # Monitor the full disk-backed workspace, including package caches. # An unnamespaced child can still hardcode RAM-backed paths: this # guard is not a replacement for the operator's memory limit. @@ -290,16 +318,36 @@ class InProcessExecutor: break except subprocess.TimeoutExpired: self._remaining() - except TimeoutError: + except task_storage.ScratchLimit as exc: + disk_exceeded = True + failure = exc + self.stopped = True + except TimeoutError as exc: timed_out = True + failure = exc + except OSError as exc: + self.phase = "scratch guard" + failure = exc finally: with suppress(ProcessLookupError): os.killpg(proc.pid, signal.SIGKILL) proc.wait() out.seek(0) err.seek(max(0, os.fstat(err.fileno()).st_size - 16384)) + stderr = task_storage.diagnostic(err.read().decode(errors="replace"), maximum=4000) + phase = "scratch guard" if disk_exceeded else "timeout" if timed_out else self.phase + if disk_exceeded: + self.stopped = True + failed = failure or (task_storage.ScratchLimit("scratch limit") if disk_exceeded else None) + reason = ( + task_storage.exception_reason(failed) + if failed + else (f"CommandFailure: exit code {proc.returncode}" if proc.returncode else "") + ) return { - **({"status": "could_not_build", "reason": "scratch limit"} if disk_exceeded else {}), + **({"status": "could_not_build", "stopped": self.stopped} if disk_exceeded else {}), + **({"reason": reason, "phase": phase} if reason else {}), + **({"error": type(failure).__name__} if failure and not (timed_out or disk_exceeded) else {}), "isolation_mode": "task-boundary", "exit_code": proc.returncode, "timed_out": timed_out, @@ -309,7 +357,7 @@ class InProcessExecutor: "stdout_bytes": os.fstat(out.fileno()).st_size, "stderr_bytes": os.fstat(err.fileno()).st_size, "stdout": out.read(16384).decode(errors="replace"), - "stderr": err.read(16384).decode(errors="replace"), + "stderr": stderr, "truncated": os.fstat(out.fileno()).st_size > 16384 or os.fstat(err.fileno()).st_size > 16384, } @@ -345,10 +393,8 @@ class InProcessExecutor: elif recipe["recipe"] == "gradle": argv += ["assemble"] built = self._task_run(argv, self._remaining(), {"disk_bytes": task_storage.DEFAULT_SCRATCH_BYTES}) - if built.get("reason") == "scratch limit": - raise task_storage.ScratchLimit("scratch limit") - if built["exit_code"] or built["timed_out"]: - return {"status": "could_not_build", "reason": "build recipe failed: " + built["stderr"][-2000:]} + if built.get("disk_limit_exceeded") or built["exit_code"] or built["timed_out"]: + return {**built, "status": "could_not_build"} # Validate internal generated links before materializing them as # regular archive members. No host/external files may be exported. _tree(self.repo, generated_links=True) @@ -361,6 +407,7 @@ class InProcessExecutor: (bundle / "spec.json").write_text(json.dumps(spec)) task_storage.check(task_storage.WORKSPACE if task_storage.WORKSPACE.is_dir() else bundle) blob = pack_checkout(bundle) + self.phase = "result upload" URLTransport().put(self.prepared_put_url, blob, self._remaining()) return {"status": "ok", "archive_bytes": len(blob), "sha256": hashlib.sha256(blob).hexdigest()} @@ -403,7 +450,7 @@ class URLTransport: except urllib.error.HTTPError as exc: if exc.code == 404: return None - raise OSError("object read failed") from None + raise OSError(f"object read failed: HTTP {exc.code}") from None def put(self, url: str, data: bytes, timeout: float) -> None: with self.opener.open( @@ -434,10 +481,14 @@ class ObjectStoreExecutor: payload = command.validate() end = time.monotonic() + timeout_seconds self.uncertain = True + last_error = "" + phase = "result upload" while time.monotonic() < end: try: # A lost PUT acknowledgement is retried with the exact same sequence. + phase = "command upload" self.transport.put(self.command_url, payload, max(0.001, end - time.monotonic())) + phase = "result read" raw = self.transport.get(self.result_url, max(0.001, end - time.monotonic())) if raw is not None: if len(raw) > MAX_RESULT_BYTES: @@ -451,10 +502,10 @@ class ObjectStoreExecutor: self.seq, self.uncertain = command.seq, False self.isolation_mode = result.get("isolation_mode", self.isolation_mode) return result - except (OSError, TimeoutError): - pass + except (OSError, TimeoutError) as exc: + last_error = task_storage.exception_reason(exc) time.sleep(min(self.poll_seconds, max(0, end - time.monotonic()))) - raise TimeoutError("executor result deadline") + raise TimeoutError(f"executor {phase} deadline; {last_error or 'no result received'}") def serve( @@ -481,8 +532,11 @@ def serve( transport.put(result_put_url, json.dumps(result).encode(), min(10, max(0.001, end - time.monotonic()))) if result.get("stopped"): return - except (OSError, TimeoutError): - pass + except (OSError, TimeoutError) as exc: + if executor.last_result: + executor.last_result["result_upload_reason"] = task_storage.exception_reason(exc) + executor.last_result["result_upload_phase"] = "result upload" + time.sleep(min(poll_seconds, max(0, end - time.monotonic()))) raise TimeoutError("executor task deadline") @@ -534,11 +588,14 @@ def main() -> None: prepared_url = os.environ.get("PREPARED_PUT_URL") # Clearing Python's environment cannot remove initial /proc/self/environ. # Refuse a misconfigured task before any repository input is opened. - if any( - value and any(word in key.upper() for word in ("KEY", "TOKEN", "SECRET", "PASSWORD", "CREDENTIAL")) + inherited = sorted( + key for key, value in os.environ.items() - ): - raise SystemExit("executor refuses inherited credentials") + if value and any(word in key.upper() for word in ("KEY", "TOKEN", "SECRET", "PASSWORD", "CREDENTIAL")) + ) + if inherited: + # Names only, never values, so a misconfigured task can be diagnosed from the worker log. + raise SystemExit("executor refuses inherited credentials: " + ", ".join(inherited)) scratch_limit = task_storage.configure() os.environ.clear() os.environ.update(clean_environment()) diff --git a/src/openultrasast/search/probe.py b/src/openultrasast/search/probe.py index e553ca9..019ca50 100644 --- a/src/openultrasast/search/probe.py +++ b/src/openultrasast/search/probe.py @@ -22,6 +22,7 @@ from typing import Any from . import _sandbox from .oracles import BrowserExecutor +from .task_storage import exception_reason from .verify import Side, verify FAMILIES = ("sql", "path", "command", "ssrf", "xss") @@ -128,13 +129,13 @@ def probe(root: Path) -> dict[str, Any]: if side.reason.startswith("verification unavailable or refused:"): raise _sandbox.IsolationUnavailable(f"{family}/{name}: {side.reason}") except Exception as exc: - record.update(status="instrument_failure", exception=f"{type(exc).__name__}: {exc}") + record.update(status="instrument_failure", exception=exception_reason(exc)) namespaces = record["namespaces"] if namespaces and all(namespaces[name]["exit_code"] != 0 for name in ("user", "pid", "net", "ipc", "uts", "mount")): record["isolation_mode"] = "task-boundary" record["requires_fresh_side_tasks"] = True if record["isolation_check"] is None: - record["isolation_check"] = {"status": "instrument_failure", "exception": str(exc)} + record["isolation_check"] = {"status": "instrument_failure", "exception": exception_reason(exc)} finally: record["wall_seconds"] = time.monotonic() - start record["peak_rss_kib"] = { @@ -165,8 +166,9 @@ def main(argv: list[str] | None = None) -> int: with urllib.request.urlopen(request, timeout=30) as response: if not 200 <= response.status < 300: return 1 - except Exception: - # Presigned URL is a bearer secret; never print upload exceptions. + except Exception as exc: + record.update(status="instrument_failure", phase="result upload", reason=exception_reason(exc)) + print(json.dumps(record), flush=True) return 1 return 0 if record["status"] == "ok" else 1 diff --git a/src/openultrasast/search/task_storage.py b/src/openultrasast/search/task_storage.py index a437c8c..071ac78 100644 --- a/src/openultrasast/search/task_storage.py +++ b/src/openultrasast/search/task_storage.py @@ -3,6 +3,7 @@ from __future__ import annotations import os +import re import stat import subprocess import tempfile @@ -23,6 +24,7 @@ def environment() -> dict[str, str]: return {} root = str(WORKSPACE) return { + "HOME": root + "/home", "TMPDIR": root + "/tmp", "npm_config_cache": root + "/.npm", "MAVEN_OPTS": "-Dmaven.repo.local=" + root + "/.m2", @@ -39,6 +41,7 @@ def configure() -> int: env = environment() if env: Path(env["TMPDIR"]).mkdir(parents=True, exist_ok=True) + Path(env["HOME"]).mkdir(parents=True, exist_ok=True) os.environ.update(env) # tempfile may have cached /tmp before this entry point ran. tempfile.tempdir = env["TMPDIR"] @@ -53,7 +56,8 @@ def check(root: Path | None = None, limit: int | None = None, *, additional_byte raise OSError("scratch root unavailable") def unreadable(error: OSError) -> None: - raise error + if not isinstance(error, FileNotFoundError): + raise error total = additional_bytes if total > limit: @@ -91,3 +95,25 @@ def communicate(proc: subprocess.Popen[Any], data: bytes, timeout: float, root: first = False if time.monotonic() >= end: raise + + +def diagnostic(value: object, *, maximum: int = 300, private: tuple[str, ...] = ()) -> str: + """Sanitize before truncation so a clipped secret cannot escape redaction.""" + text = str(value) + for key, secret in os.environ.items(): + if len(secret) >= 4 and any(word in key.upper() for word in ("KEY", "TOKEN", "SECRET", "PASSWORD", "CREDENTIAL")): + text = text.replace(secret, "[redacted]") + text = re.sub(r"(?i)bearer\s+[^\s,;\"']+", "Bearer [redacted]", text) + text = re.sub(r"(?i)(?:https?|s3)://[^\s<>\"']+", "[redacted-url]", text) + text = re.sub( + r"(?i)([\w-]*(?:authorization|signature|credential|token|secret|password|key))\s*[:=]\s*[^\s,;&\"']+", r"\1=[redacted]", text + ) + for name in sorted(set(private), key=len, reverse=True): + if name: + text = text.replace(name, "[repository]") + return "\n".join(text.splitlines()[-20:])[-maximum:] + + +def exception_reason(exc: BaseException, *, private: tuple[str, ...] = ()) -> str: + prefix = type(exc).__name__ + ": " + return prefix + diagnostic(exc, maximum=max(0, 300 - len(prefix)), private=private) diff --git a/src/openultrasast/search/verify.py b/src/openultrasast/search/verify.py index f94fde0..4939472 100644 --- a/src/openultrasast/search/verify.py +++ b/src/openultrasast/search/verify.py @@ -29,6 +29,7 @@ from ..sandbox import SandboxJob, SandboxResult from . import _sandbox, task_storage from .demo import BUILD_RECIPES, RUNTIMES, capture_output, load_demo, substitute from .oracles import BrowserExecutor, Canary, Oracle, oracle_for +from .task_storage import exception_reason @dataclass(frozen=True) @@ -414,10 +415,10 @@ def _side(side: Side, demo: Path, family: str, timeout: int, browser: BrowserExe finally: oracle.close() return record("observed", "fresh task run completed" if task_boundary else "three fresh runs completed") - except task_storage.ScratchLimit: - return record("could_not_build", "scratch limit") + except task_storage.ScratchLimit as exc: + return record("could_not_build", exception_reason(exc)) except (OSError, ValueError, AssertionError, KeyError, IndexError, TypeError) as exc: - return record("could_not_run", f"verification unavailable or refused: {exc}") + return record("could_not_run", "verification unavailable or refused: " + exception_reason(exc)) def _run_task( @@ -508,7 +509,7 @@ def verify( try: load_demo(artefact) except (OSError, ValueError) as exc: - return VerificationRecord("inconclusive", (), time.monotonic() - started, "invalid declarative demo: " + str(exc)) + return VerificationRecord("inconclusive", (), time.monotonic() - started, "invalid declarative demo: " + exception_reason(exc)) mode = "task-boundary" if task_dispatcher is not None else _sandbox.isolation_mode() if task_dispatcher is None: try: diff --git a/src/openultrasast/search/verify_task.py b/src/openultrasast/search/verify_task.py index d4bca1c..dac65bc 100644 --- a/src/openultrasast/search/verify_task.py +++ b/src/openultrasast/search/verify_task.py @@ -212,8 +212,8 @@ def task_main() -> None: demo.mkdir() (demo / "demo.json").write_text(json.dumps(spec["demo"])) result = run_side(Side(bundle / "checkout", products=bundle / "products"), demo, spec["family"], spec["timeout_seconds"]) - except task_storage.ScratchLimit: - result = SideRecord("could_not_build", (), 0, "scratch limit", isolation_mode="task-boundary") + except task_storage.ScratchLimit as exc: + result = SideRecord("could_not_build", (), 0, task_storage.exception_reason(exc), isolation_mode="task-boundary") upload(urls["RESULT_URL"], json.dumps(asdict(result)).encode()) diff --git a/src/openultrasast/search/worker.py b/src/openultrasast/search/worker.py index 94ab32e..7042f27 100644 --- a/src/openultrasast/search/worker.py +++ b/src/openultrasast/search/worker.py @@ -20,6 +20,7 @@ from . import _sandbox from .coordinator import SearchTask from .demo import DEMO_SCHEMA, load_demo, validate_demo from .executor import ObjectStoreExecutor +from .task_storage import exception_reason def _schema(name: str, description: str, properties: dict[str, Any], required: list[str]) -> dict[str, Any]: @@ -286,7 +287,7 @@ class Worker: self.evidence.add(ref) result["evidence_ref"] = ref except (ValueError, OSError, KeyError, TypeError) as exc: - result = {"error": type(exc).__name__} + result = {"error": type(exc).__name__, "reason": exception_reason(exc)} messages.append({"role": "tool", "tool_call_id": call["id"], "content": json.dumps(result)}) return {"status": "execution_failure", "failure": "tool step limit exceeded"} @@ -307,18 +308,18 @@ class Worker: "cost_usd": meter.spent, "isolation_mode": getattr(self.executor, "isolation_mode", "brain-only"), } - except _sandbox.IsolationUnavailable: + except _sandbox.IsolationUnavailable as exc: return { "status": "instrument_failure", - "failure": "sandbox isolation unavailable", + "failure": exception_reason(exc), "cost_usd": meter.spent, "isolation_mode": getattr(self.executor, "isolation_mode", "brain-only"), } except Exception as exc: - # Provider exceptions can contain keys/URLs. Never persist their text. + # Preserve the diagnostic after removing credentials and URLs. return { "status": "execution_failure", - "failure": type(exc).__name__, + "failure": exception_reason(exc), "cost_usd": meter.spent, "isolation_mode": getattr(self.executor, "isolation_mode", "brain-only"), } diff --git a/tests/test_ax_batch.py b/tests/test_ax_batch.py index cbd05ff..94e902f 100644 --- a/tests/test_ax_batch.py +++ b/tests/test_ax_batch.py @@ -274,13 +274,13 @@ def test_capacity_backpressure_retries_same_manifest_until_deadline(tmp_path, mo ) assert lane(item(), tmp_path, 0)["status"] == "ok" assert len(applies) >= 3 and len(set(applies)) == 1 - assert len(deletes) == 1 + assert len(deletes) == len(applies) busy[0] = True lane.pause = lambda seconds: now.__setitem__(0, now[0] + seconds) deletes.clear() result = lane(item(), tmp_path, 1) assert result["status"] == "capacity_timeout" - assert not deletes + assert deletes calls = [] def no_vm(*args): diff --git a/tests/test_search_executor.py b/tests/test_search_executor.py index 41c85c5..e4b0b79 100644 --- a/tests/test_search_executor.py +++ b/tests/test_search_executor.py @@ -39,7 +39,7 @@ def test_paths_and_oversize(tmp_path): def test_environment_allowlist(): assert clean_environment( {"OPENROUTER_API_KEY": "secret", "GEMINI_API_KEY": "secret", "PATH": "/evil", "AWS_SECRET_ACCESS_KEY": "secret"} - ) == {"PATH": "/usr/bin:/bin", "LANG": "C.UTF-8"} + ) == {"PATH": "/venv/bin:/opt/java/openjdk/bin:/usr/local/bin:/usr/bin:/bin", "LANG": "C.UTF-8", "JAVA_HOME": "/opt/java/openjdk"} class Mailbox: @@ -240,3 +240,79 @@ def test_sandbox_receives_exact_remaining_wall_limit(tmp_path, monkeypatch): InProcessExecutor(tmp_path).submit("run", {"command": ["/bin/true"]}, timeout_seconds=0.2) assert 0 < seen["probe"] <= 0.2 assert 0 < seen["wall"] <= 0.2 + + +def test_spawn_failure_has_diagnostic(tmp_path): + result = InProcessExecutor(tmp_path, task_boundary=True).execute(Command(1, "run", {"command": ["/missing-executor-program"]})) + assert result["phase"] == "spawn" + assert result["exit_code"] is None + assert "FileNotFoundError:" in result["reason"] + assert "No such file" in result["reason"] + + +def test_timeout_preserves_child_diagnostics(tmp_path): + import sys + + result = InProcessExecutor(tmp_path, task_boundary=True).execute( + Command( + 1, + "run", + {"command": [sys.executable, "-c", "import sys,time; print('before timeout',file=sys.stderr,flush=True); time.sleep(10)"]}, + 0.3, + ) + ) + assert result["phase"] == "timeout" + assert result["exit_code"] is not None + assert result["stderr"] == "before timeout" + assert result["reason"].startswith("TimeoutError:") + + +def test_upload_failure_reports_phase(tmp_path, monkeypatch): + app = InProcessExecutor(tmp_path, task_boundary=True) + + def upload(command): + app.phase = "result upload" + raise OSError("upload HTTP 403 https://host/path?X-Amz-Signature=secret") + + monkeypatch.setattr(app, "_execute", upload) + result = app.execute(Command(1, "prepare_verification", {})) + assert result["phase"] == "result upload" + assert "HTTP 403" in result["reason"] + assert "secret" not in result["reason"] + + +def test_diagnostics_sanitize_before_tail_and_keep_twenty_lines(monkeypatch): + from openultrasast.search.task_storage import diagnostic, exception_reason + + monkeypatch.setenv("SERVICE_TOKEN", "environment-secret") + detail = ( + "https://host/repo?X-Amz-Signature=url-secret Authorization: Bearer bearer-secret X-Amz-Signature=bare-secret environment-secret" + ) + result = exception_reason(ValueError(detail)) + assert result.startswith("ValueError:") and len(result) <= 300 + for secret in ("url-secret", "bearer-secret", "bare-secret", "environment-secret", "host/repo"): + assert secret not in result + assert len(diagnostic("\n".join(str(i) for i in range(40)), maximum=4000).splitlines()) == 20 + assert len(exception_reason(ValueError("x" * 1000))) <= 300 + + +def test_guard_io_failure_after_child_exit_is_not_success(tmp_path, monkeypatch): + import sys + + from openultrasast.search import task_storage + + checks = [] + original = task_storage.check + + def check(*args, **kwargs): + checks.append(1) + if len(checks) >= 3: + raise PermissionError("cache cannot be read") + return original(*args, **kwargs) + + monkeypatch.setattr(task_storage, "check", check) + result = InProcessExecutor(tmp_path, task_boundary=True).execute(Command(1, "run", {"command": [sys.executable, "-c", "pass"]})) + assert result["error"] == "PermissionError" + assert result["phase"] == "scratch guard" + assert result["exit_code"] is not None + assert "cache cannot be read" in result["reason"] diff --git a/tests/test_search_executor_smoke.py b/tests/test_search_executor_smoke.py index c0379e0..9d81e31 100644 --- a/tests/test_search_executor_smoke.py +++ b/tests/test_search_executor_smoke.py @@ -142,9 +142,9 @@ def test_docker_managed_task_uses_docker_and_no_heap_flag(tmp_path, monkeypatch) ) record = smoke.run_smoke(lane, "https://github.com/example/project", "a" * 40) assert record["status"] == "ok" - assert [a[1] for a in calls] == ["run", "rm"] - assert "--heap-profile" not in calls[0] - assert calls[1][2] == "-f" + assert [a[1] for a in calls] == ["image", "run", "rm"] + assert "--heap-profile" not in calls[1] + assert calls[2][2] == "-f" def test_apply_exception_still_deletes(tmp_path, monkeypatch): @@ -160,3 +160,104 @@ def test_apply_exception_still_deletes(tmp_path, monkeypatch): assert result["status"] == "instrument_failure" assert events == ["apply", "delete"] assert "secret.example" not in json.dumps(result) + + +def test_smoke_retains_sanitized_command_failure(tmp_path, monkeypatch): + executor, lane, events = setup(tmp_path, monkeypatch) + original = executor._task_run + + def fail(argv, timeout, limits): + if "pip" in argv: + return { + "exit_code": 17, + "stderr": "project failed: Bearer hidden https://host/project?q=secret", + "stdout": "", + "phase": "run", + "reason": "CommandFailure: exit code 17", + } + return original(argv, timeout, limits) + + monkeypatch.setattr(executor, "_task_run", fail) + record = smoke.run_smoke(lane, "https://github.com/example/project", "a" * 40) + row = record["commands"][-1] + assert row["exit_code"] == 17 and row["phase"] == "run" + assert "CommandFailure" in row["reason"] + assert all(word not in json.dumps(record) for word in ("hidden", "secret", "project", "example")) + assert events == ["apply", "delete"] + + +def test_docker_missing_image_is_diagnosable_without_pull(tmp_path, monkeypatch): + import subprocess + + monkeypatch.setenv("S3_ENDPOINT", "https://files.example") + calls = [] + + def runner(argv, **kwargs): + calls.append(argv) + return subprocess.CompletedProcess(argv, 1, "", "No such image: pinned-image") + + lane = batch.DockerLane( + SimpleNamespace(atespace="default"), + batch.Workload("image@sha256:" + "a" * 64, frozenset(), 900, None), + store=Store(Executor(tmp_path, task_boundary=True)), + runner=runner, + ) + record = smoke.run_smoke(lane, "https://github.com/example/project", "a" * 40) + assert record["status"] == "instrument_failure" + assert "ValueError:" in record["reason"] and "No such image" in record["reason"] + assert [c[1] for c in calls] == ["image"] + + +@pytest.mark.parametrize("delete_fails", [False, True]) +def test_capacity_partial_create_deleted_before_retry(tmp_path, monkeypatch, delete_fails): + executor, lane, events = setup(tmp_path, monkeypatch) + live, attempts = set(), [] + + def ax(*args, **kwargs): + events.append(args[0]) + if args[0] == "apply": + assert not live + live.add(kwargs["manifest"]["metadata"]["name"]) + attempts.append(1) + if len(attempts) == 1: + raise batch.NoFreeWorkers("partially provisioned") + elif args[0] == "delete": + if delete_fails: + raise OSError("cannot delete") + live.discard(args[2]) + + lane.ax, lane.pause = ax, lambda seconds: None + record = smoke.run_smoke(lane, "https://github.com/example/project", "a" * 40) + assert len(attempts) == (1 if delete_fails else 2) + assert record["status"] == ("instrument_failure" if delete_fails else "ok") + if not delete_fails: + assert events == ["apply", "delete", "apply", "delete"] and not live + else: + # Simulated leaked task deliberately retains its reservation; restore test global. + batch.AX_SLOTS.release() + + +def test_apply_timeout_does_not_resubmit(tmp_path, monkeypatch): + import subprocess + + executor, lane, events = setup(tmp_path, monkeypatch) + + def ax(*args, **kwargs): + events.append(args[0]) + if args[0] == "apply": + raise subprocess.TimeoutExpired("apply", 30) + + lane.ax = ax + record = smoke.run_smoke(lane, "https://github.com/example/project", "a" * 40) + assert events == ["apply", "delete"] + assert "TimeoutExpired:" in record["reason"] + + +def test_dispatch_rejects_duplicate_items_before_launch(tmp_path): + item = batch.Item("same", {}, "result.json", (), {}, "") + + def unexpected(*args): + pytest.fail("duplicate inputs must be rejected before any dispatch") + + with pytest.raises(ValueError, match="duplicate dispatch item"): + batch.dispatch([item, item], batch.Workload("image", frozenset(), 1, None), tmp_path, {"ax": unexpected}, lanes=("ax",)) diff --git a/tests/test_search_task_storage.py b/tests/test_search_task_storage.py index 3f5c389..3a45e4f 100644 --- a/tests/test_search_task_storage.py +++ b/tests/test_search_task_storage.py @@ -61,7 +61,7 @@ def test_executor_checks_caches_and_stops_at_scratch_limit(workspace, monkeypatc "p=Path(os.environ['PIP_CACHE_DIR']); p.mkdir(); (p/'large').write_bytes(b'x'*30000); " + ("time.sleep(10)" if wait else "") ) result = app.submit("run", {"command": [sys.executable, "-c", code]}, timeout_seconds=3) - assert result["status"] == "could_not_build" and result["reason"] == "scratch limit" + assert result["status"] == "could_not_build" and result["reason"] == "ScratchLimit: scratch limit" assert result["stopped"] and app.stopped @@ -126,7 +126,7 @@ def test_verifier_reports_scratch_limit(workspace, monkeypatch): (demo / "demo.json").write_text(json.dumps(spec()["demo"])) monkeypatch.setenv("OUSAST_SCRATCH_BYTES", "25000") result = verify_side_task(Side(repo), demo, "path") - assert result.outcome == "could_not_build" and result.reason == "scratch limit", result + assert result.outcome == "could_not_build" and result.reason == "ScratchLimit: scratch limit", result def test_preparation_builds_before_export_and_exports_products(workspace, monkeypatch): @@ -165,9 +165,60 @@ def test_open_logs_count_toward_scratch_guard(workspace, monkeypatch): monkeypatch.setenv("OUSAST_SCRATCH_BYTES", "20000") app = executor.InProcessExecutor(repo, task_boundary=True) result = app.submit("run", {"command": [sys.executable, "-c", "print('x'*30000)"]}) - assert result["status"] == "could_not_build" and result["reason"] == "scratch limit" + assert result["status"] == "could_not_build" and result["reason"] == "ScratchLimit: scratch limit" def test_unreadable_scratch_is_not_reported_as_empty(tmp_path): with pytest.raises(OSError, match="scratch root unavailable"): task_storage.check(tmp_path / "missing") + + +def test_pip_removing_enumerated_directory_is_not_guard_failure(workspace, monkeypatch): + original = task_storage.os.walk + + def disappearing(root, **kwargs): + kwargs["onerror"](FileNotFoundError(2, "removed pip build directory", str(root / "pip-build"))) + yield from original(root, **kwargs) + + monkeypatch.setattr(task_storage.os, "walk", disappearing) + (workspace / "input").write_bytes(b"input") + assert task_storage.check(workspace) >= 5 + + +def test_guard_still_refuses_permission_errors(workspace, monkeypatch): + def unreadable(root, **kwargs): + kwargs["onerror"](PermissionError("permission denied")) + return iter(()) + + monkeypatch.setattr(task_storage.os, "walk", unreadable) + with pytest.raises(PermissionError): + task_storage.check(workspace) + + +def test_child_environment_has_disk_home_and_owned_runtime_path(workspace): + repo = workspace / "checkout" + repo.mkdir() + (repo / "input").write_bytes(b"proof") + code = ( + "import os; from pathlib import Path; print(Path('input').stat().st_size); " + "print(os.environ['PATH']); p=Path.home()/'probe'; p.write_text('ok'); print(p)" + ) + result = executor.InProcessExecutor(repo, task_boundary=True).submit("run", {"command": [sys.executable, "-c", code]}) + assert result["exit_code"] == 0 + assert result["stdout"].startswith("5\n/venv/bin:") + assert (workspace / "home/probe").read_text() == "ok" + + +def test_scratch_failure_preserves_exit_and_stderr(workspace, monkeypatch): + repo = workspace / "checkout" + repo.mkdir() + monkeypatch.setenv("OUSAST_SCRATCH_BYTES", "20000") + code = ( + "import sys,time; from pathlib import Path; print('pip build detail',file=sys.stderr,flush=True); " + "Path('big').write_bytes(b'x'*30000); time.sleep(10)" + ) + result = executor.InProcessExecutor(repo, task_boundary=True).submit("run", {"command": [sys.executable, "-c", code]}) + assert result["phase"] == "scratch guard" + assert result["exit_code"] is not None + assert result["stderr"] == "pip build detail" + assert result["reason"].startswith("ScratchLimit:") diff --git a/tests/test_search_worker.py b/tests/test_search_worker.py index 6aed1a7..31b6851 100644 --- a/tests/test_search_worker.py +++ b/tests/test_search_worker.py @@ -113,7 +113,7 @@ def test_reservation_precedes_each_call_and_refuses_over_budget(tmp_path): job = task() job.limits["max_call_usd"] = 0.000001 result = worker(job) - assert result["failure"] == "BudgetExhausted" + assert result["failure"].startswith("BudgetExhausted: ") assert len(client.calls) == 1