From 939c297e6e465ccad95d1a11b74854ae0ffda39b Mon Sep 17 00:00:00 2001 From: Anton Razzhigaev Date: Thu, 27 Aug 2026 02:24:48 +0000 Subject: [PATCH] Harden CyberGym verifier response handling Co-authored-by: Ouroboros <311266734+ouroboros-agent@users.noreply.github.com> --- .../benchmarks/cybergym/cybergym_executor.py | 188 +++++++++++++++--- tests/test_cybergym_executor.py | 119 +++++++++++ 2 files changed, 275 insertions(+), 32 deletions(-) diff --git a/devtools/benchmarks/cybergym/cybergym_executor.py b/devtools/benchmarks/cybergym/cybergym_executor.py index 45b6e7270..421d3f59e 100644 --- a/devtools/benchmarks/cybergym/cybergym_executor.py +++ b/devtools/benchmarks/cybergym/cybergym_executor.py @@ -433,12 +433,30 @@ def _runtime_value(payload: Mapping[str, Any], *keys: str) -> Any: return None -def _unwrap_http_json(value: Any, *, operation: str) -> Mapping[str, Any]: - """Normalize an injected HTTP response and reject transport/body errors.""" +_HTTP_BODY_MISSING = object() + +def _unwrap_http_payload( + value: Any, *, operation: str, allow_list: bool = False +) -> Mapping[str, Any] | list[Any]: + """Normalize an injected HTTP response and reject transport/body errors. + + ``urllib_json`` returns the decoded upstream body directly, while unit and + alternate transports may return an envelope such as + ``{"status_code": 200, "body": ...}``. Keep both forms equivalent and + never turn an HTTP/body error into an empty result that could be mistaken + for a legitimate verifier response. + """ + + if isinstance(value, list): + if allow_list: + return value + raise ExecutorFailure(f"{operation} returned a list where an object was required") if not isinstance(value, Mapping): raise ExecutorFailure(f"{operation} returned a non-object response") - status_code = value.get("status_code", value.get("http_status")) + + envelope = value + status_code = envelope.get("status_code", envelope.get("http_status")) if status_code is not None: try: status = int(status_code) @@ -446,13 +464,41 @@ def _unwrap_http_json(value: Any, *, operation: str) -> Mapping[str, Any]: raise ExecutorFailure(f"{operation} returned an invalid HTTP status") from exc if status != 200: raise ExecutorFailure(f"{operation} returned HTTP {status}") - body = value.get("body") - if isinstance(body, Mapping): - value = body - error = value.get("error") + + error = envelope.get("error") if error not in (None, "", False, {}): raise ExecutorFailure(f"{operation} returned an error object") - return value + if envelope.get("ok") is False: + raise ExecutorFailure(f"{operation} returned an unsuccessful response") + + body = envelope.get("body", _HTTP_BODY_MISSING) + if body is not _HTTP_BODY_MISSING: + if isinstance(body, Mapping): + value = body + elif isinstance(body, list) and allow_list: + return body + else: + raise ExecutorFailure(f"{operation} returned an invalid response body") + + if isinstance(value, Mapping): + error = value.get("error") + if error not in (None, "", False, {}): + raise ExecutorFailure(f"{operation} returned an error object") + if value.get("ok") is False: + raise ExecutorFailure(f"{operation} returned an unsuccessful response") + return value + if isinstance(value, list) and allow_list: + return value + raise ExecutorFailure(f"{operation} returned an invalid response body") + + +def _unwrap_http_json(value: Any, *, operation: str) -> Mapping[str, Any]: + """Normalize an injected HTTP response that must contain an object.""" + + payload = _unwrap_http_payload(value, operation=operation, allow_list=False) + if not isinstance(payload, Mapping): # defensive; the helper already checks + raise ExecutorFailure(f"{operation} returned a non-object response") + return payload def _positive_int(value: Any, field: str) -> int: @@ -467,6 +513,15 @@ def _positive_int(value: Any, field: str) -> int: return number +def _require_exact_effort(value: Any) -> str: + """Accept only the owner-approved literal reasoning effort ``high``.""" + + effort = str(value or "").strip() + if effort != "high": + raise ExecutorFailure("gateway result effort is not exactly high") + return effort + + def _gateway_path(base: str, path: str) -> str: return base.rstrip("/") + "/" + path.lstrip("/") @@ -568,19 +623,59 @@ def _response_task_id(response: Mapping[str, Any]) -> str: return str(value or "").strip() -def _response_hash(response: Mapping[str, Any]) -> str: - nested = response.get("response") - if isinstance(nested, Mapping): - response = {**nested, **response} - return str(response.get("poc_hash") or response.get("sha256") or response.get("hash") or "").strip().lower() - - def _record_matches(record: Mapping[str, Any], task_id: str, digest: str) -> bool: record_task = str(record.get("task_id") or "") record_hash = str(record.get("poc_hash") or record.get("hash") or "").lower() return record_task == task_id and record_hash == digest +def _response_poc_id(response: Mapping[str, Any]) -> str: + """Extract the upstream submission id without treating it as a byte hash.""" + + nested = response.get("response") + if isinstance(nested, Mapping): + response = {**nested, **response} + value = response.get("poc_id") or response.get("submission_id") + text = str(value or "").strip() + if not text or len(text) > 256 or any(char.isspace() or ord(char) < 32 for char in text): + raise ExecutorFailure("official submit response omitted a valid poc_id") + return text + + +def _validate_verify_response( + value: Any, *, expected_poc_id: str = "" +) -> Mapping[str, Any]: + """Validate the pinned ``/verify-agent-pocs`` response shape. + + The upstream endpoint returns ``{"message": str, "poc_ids": [str, ...]}`` + with HTTP 200. A successful transport carrying an empty/malformed body is + not evidence that verification happened, so fail closed before querying + records. ``expected_poc_id`` binds the response to the designated final + submission while preserving all raw exit codes in the later DB record. + """ + + response = _unwrap_http_json(value, operation="CyberGym verify-agent-pocs") + message = response.get("message") + if not isinstance(message, str) or not message.strip(): + raise ExecutorFailure("verify-agent-pocs response omitted its message") + raw_ids = response.get("poc_ids") + if not isinstance(raw_ids, Sequence) or isinstance(raw_ids, (str, bytes)): + raise ExecutorFailure("verify-agent-pocs response omitted its poc_ids list") + poc_ids: list[str] = [] + for raw_id in raw_ids: + if not isinstance(raw_id, str): + raise ExecutorFailure("verify-agent-pocs response contains a non-string poc_id") + poc_id = raw_id.strip() + if not poc_id or len(poc_id) > 256 or any(char.isspace() or ord(char) < 32 for char in poc_id): + raise ExecutorFailure("verify-agent-pocs response contains an invalid poc_id") + poc_ids.append(poc_id) + if not poc_ids: + raise ExecutorFailure("verify-agent-pocs response contains no verified poc_ids") + if expected_poc_id and expected_poc_id not in poc_ids: + raise ExecutorFailure("verify-agent-pocs response omitted the designated poc_id") + return response + + class CyberGymExecutor: """Run one task at a time against a campaign-owned sidecar.""" @@ -905,7 +1000,7 @@ class CyberGymExecutor: paths = response.get("paths") if not isinstance(paths, Mapping): raise ExecutorFailure("CyberGym readiness response has no OpenAPI paths") - required = {"/submit-vul", "/query-poc", "/verify-agent-pocs"} + required = {"/submit-vul", "/submit-fix", "/query-poc", "/verify-agent-pocs"} if not required.issubset(paths): raise ExecutorFailure("CyberGym readiness response misses a required route") return @@ -1106,9 +1201,14 @@ class CyberGymExecutor: raise ExecutorFailure("official submit.sh response omitted its masked task id") if declared_masked_id and declared_masked_id != masked_id: raise ExecutorFailure("submit response task id conflicts with generated script") - response_hash = _response_hash(response) - if response_hash and response_hash != marker.sha256: - raise ExecutorFailure("submit response hash does not match final.poc") + # The pinned upstream /submit-vul response has no PoC hash. Its + # ``poc_id`` is the submission identity; the bytes are bound by our + # local marker and the later protected query record. Do not infer a + # hash from incidental ``hash``/``sha256`` fields in an alternate + # response body, which made a valid nonzero vulnerable exit look like + # a transport failure. + poc_id = _response_poc_id(response) + response["poc_id"] = poc_id response["final_poc_sha256"] = marker.sha256 response["masked_task_id"] = masked_id response["submit_returncode"] = result.returncode @@ -1120,19 +1220,31 @@ class CyberGymExecutor: # The private server is loopback-published; do not put the key in any # checkpoint or returned artifact. plan = self._network_plan("campaign") - payload = self.config.http_runner( - "POST", f"http://127.0.0.1:{plan.verifier_host_port}/query-poc", - body={"agent_id": agent_id, "task_id": real_task_id}, headers=headers, timeout=60, + payload = _unwrap_http_payload( + self.config.http_runner( + "POST", f"http://127.0.0.1:{plan.verifier_host_port}/query-poc", + body={"agent_id": agent_id, "task_id": real_task_id}, headers=headers, timeout=60, + ), + operation="CyberGym private query", + allow_list=True, ) # The pinned upstream route returns a bare JSON list. A few private # proxies wrap it in ``records``/``items``; accept both shapes without # weakening the task/hash binding below. - records = payload if isinstance(payload, list) else ( - payload.get("records", payload.get("pocs", payload)) if isinstance(payload, Mapping) else [] - ) - if isinstance(records, Mapping): - records = records.get("items", []) - return [dict(item) for item in records if isinstance(item, Mapping)] if isinstance(records, Sequence) and not isinstance(records, (str, bytes)) else [] + if isinstance(payload, list): + records: Any = payload + else: + records = payload.get("records", payload.get("pocs", payload.get("items"))) + if not isinstance(records, Sequence) or isinstance(records, (str, bytes)): + raise ExecutorFailure("CyberGym private query returned no records list") + normalized: list[dict[str, Any]] = [] + for item in records: + if not isinstance(item, Mapping): + raise ExecutorFailure("CyberGym private query returned a malformed record") + normalized.append(dict(item)) + if not normalized: + raise ExecutorFailure("CyberGym private query returned no records") + return normalized def run_task(self, task: TaskSpec, task_dir: pathlib.Path) -> Mapping[str, Any]: """Execute one admitted task; callback-compatible with ``run_campaign``.""" @@ -1168,9 +1280,20 @@ class CyberGymExecutor: # verify-agent-pocs is the upstream operation that reruns both images. key = self._ensure_key() plan_host = f"http://127.0.0.1:{plan.verifier_host_port}" - verify_response = self.config.http_runner("POST", plan_host + "/verify-agent-pocs", body={"agent_id": agent_id}, headers={"X-API-Key": key}, timeout=300) + submitted_poc_id = _response_poc_id(submit_response) + verify_response = _validate_verify_response( + self.config.http_runner( + "POST", plan_host + "/verify-agent-pocs", + body={"agent_id": agent_id}, headers={"X-API-Key": key}, timeout=300, + ), + expected_poc_id=submitted_poc_id, + ) records = self._private_query(agent_id, task.task_id) - matching = [item for item in records if _record_matches(item, task.task_id, digest)] + matching = [ + item for item in records + if _record_matches(item, task.task_id, digest) + and str(item.get("poc_id") or "").strip() == submitted_poc_id + ] if not matching: raise ExecutorFailure("private query returned no record for the designated final PoC") record = matching[-1] @@ -1193,8 +1316,9 @@ class CyberGymExecutor: completion_tokens = _runtime_value(gateway_result, "completion_tokens", "output_tokens", "tokens_out") if observed_model != self.config.model: raise ExecutorFailure("gateway result omitted or changed the exact requested model") - if not observed_provider or not observed_effort: - raise ExecutorFailure("gateway result omitted provider or effort telemetry") + if not observed_provider: + raise ExecutorFailure("gateway result omitted provider telemetry") + observed_effort = _require_exact_effort(observed_effort) _positive_int(prompt_tokens, "gateway prompt_tokens") _positive_int(completion_tokens, "gateway completion_tokens") trial = { diff --git a/tests/test_cybergym_executor.py b/tests/test_cybergym_executor.py index 905239b9a..81ede4a38 100644 --- a/tests/test_cybergym_executor.py +++ b/tests/test_cybergym_executor.py @@ -7,6 +7,8 @@ the live smoke remains an operator action documented by the benchmark. from __future__ import annotations +import hashlib +import json import pathlib import pytest @@ -16,6 +18,8 @@ from devtools.benchmarks.cybergym.cybergym_executor import ( CyberGymExecutor, ExecutorConfig, ExecutorFailure, + _require_exact_effort, + _validate_verify_response, ) @@ -99,6 +103,7 @@ def test_start_uses_same_absolute_server_root_and_docs_probe(tmp_path, monkeypat "openapi": "3.0.0", "paths": { "/submit-vul": {}, + "/submit-fix": {}, "/query-poc": {}, "/verify-agent-pocs": {}, }, @@ -110,6 +115,120 @@ def test_start_uses_same_absolute_server_root_and_docs_probe(tmp_path, monkeypat assert seen_http == [("GET", "http://127.0.0.1:8667/openapi.json")] +def test_readiness_rejects_openapi_without_private_submit_fix(tmp_path, monkeypatch): + config = _config(tmp_path) + monkeypatch.setenv("CYBERGYM_API_KEY", "test-secret-value") + import devtools.benchmarks.cybergym.cybergym_executor as executor_module + + ticks = iter((0.0, 121.0)) + monkeypatch.setattr(executor_module.time, "monotonic", lambda: next(ticks)) + + def command(argv, *, cwd=None, env=None, timeout=None): + if "network" in argv and "create" in argv: + return CommandResult(0, "network-id\n", "") + if "inspect" in argv and "network" in argv: + return CommandResult( + 0, + '[{"Name":"cybergym-internal","Id":"network-id","Internal":true,"Driver":"bridge","Labels":{"com.ouroboros.campaign":"test-campaign"}}]', + "", + ) + if "run" in argv: + return CommandResult(0, "server-container-id\n", "") + if "inspect" in argv and "container" in argv: + return CommandResult( + 0, + '[{"Name":"/cybergym-server-test-campaign","Id":"server-container-id","State":{"Running":true},"HostConfig":{"NetworkMode":"cybergym-internal"},"NetworkSettings":{"Networks":{"cybergym-internal":{"Aliases":["cybergym-server-test-campaign"],"NetworkID":"network-id"}}},"Config":{"Labels":{},"Image":"sha256:' + + "1" * 64 + + '"}}]', + "", + ) + return CommandResult(0, "", "") + + def http(method, url, **kwargs): + return {"openapi": "3.0.0", "paths": {"/submit-vul": {}, "/query-poc": {}, "/verify-agent-pocs": {}}} + + executor = CyberGymExecutor( + dataclasses_replace(config, command_runner=command, http_runner=http, provider_probe=False) + ) + with pytest.raises(ExecutorFailure, match="documented route"): + executor.start() + + +def test_verify_response_requires_success_body_and_designated_poc(): + good = {"message": "All 1 PoCs for this agent_id have been verified", "poc_ids": ["poc-1"]} + assert _validate_verify_response(good, expected_poc_id="poc-1") == good + with pytest.raises(ExecutorFailure, match="HTTP 500"): + _validate_verify_response({"status_code": 500, "body": {"detail": "failed"}}) + with pytest.raises(ExecutorFailure, match="poc_ids"): + _validate_verify_response({"message": "ok", "poc_ids": []}) + with pytest.raises(ExecutorFailure, match="designated poc_id"): + _validate_verify_response(good, expected_poc_id="other") + + +def test_observed_effort_must_be_exactly_high(): + assert _require_exact_effort("high") == "high" + for value in ("", "High", "max", None): + with pytest.raises(ExecutorFailure, match="exactly high"): + _require_exact_effort(value) + + +def test_private_query_rejects_http_and_body_errors(tmp_path, monkeypatch): + config = _config(tmp_path) + monkeypatch.setenv("CYBERGYM_API_KEY", "test-secret-value") + executor = CyberGymExecutor( + dataclasses_replace( + config, + http_runner=lambda *args, **kwargs: {"status_code": 404, "body": {"detail": "Record not found"}}, + ) + ) + with pytest.raises(ExecutorFailure, match="HTTP 404"): + executor._private_query("agent-" + "a" * 24, "arvo:1") + + executor = CyberGymExecutor( + dataclasses_replace( + config, + http_runner=lambda *args, **kwargs: {"status_code": 200, "body": {"error": {"message": "bad"}}}, + ) + ) + with pytest.raises(ExecutorFailure, match="error object"): + executor._private_query("agent-" + "a" * 24, "arvo:1") + + +def test_submit_response_binds_poc_id_not_nonexistent_hash_and_keeps_exit_code(tmp_path): + config = _config(tmp_path) + task_dir = config.run_root / "task" + task_dir.mkdir() + (task_dir / "final.poc").write_bytes(b"poc-bytes") + (task_dir / "submit.sh").write_text("TASK_ID=opaque1234\n", encoding="utf-8") + + def command(argv, *, cwd=None, env=None, timeout=None): + return CommandResult( + 0, + json.dumps( + { + "task_id": "opaque1234", + "poc_id": "poc-1", + "exit_code": 71, + "output": "known", + # Upstream does not define a response hash; an incidental + # field must not override the local marker binding. + "hash": "not-the-poc-hash", + } + ), + "", + ) + + response, digest, masked = CyberGymExecutor( + dataclasses_replace(config, command_runner=command) + )._submit_final( # noqa: SLF001 - boundary contract assertion + type("Task", (), {"task_id": "arvo:1"})(), task_dir, "workspace" + ) + assert response["poc_id"] == "poc-1" + assert response["exit_code"] == 71 + assert digest == hashlib.sha256(b"poc-bytes").hexdigest() + assert masked == "opaque1234" + + def dataclasses_replace(config, **changes): import dataclasses