PROD-250: fix(core): reject malformed governance responses

README.md

@@ -39,6 +39,18 @@ uv sync --extra http    # + HTTP instrumentation targets
 uv sync --extra db      # + DB instrumentation targets
 ```
 
+## Sandbox execution hook contract
+
+`sandbox_execution` hook spans carry privacy-allowlisted, bounded evidence in
+`attributes` only. They must not include raw command output, environment
+variables, credentials, or body-like root fields. Direct hooks may identify the
+sandbox template with `openbox.sandbox.compatibility_id` and the lowercase,
+64-character SHA-256 attribute `openbox.sandbox.template_sha256`, alongside the
+other allowlisted sandbox evidence attributes.
+
+Sandbox hook governance uses the SDK's existing direct evaluate path and does
+not require an OTLP exporter.
+
 ## Import safety
 
 `openbox_core.__init__` and all `openbox_core.contracts.*` modules import

openbox_core/client.py

@@ -9,8 +9,9 @@
 - Signed requests send ``content=body_bytes`` — NEVER ``json=`` (client-side
   re-serialization breaks Core's body-hash verification).
 - ``httpx`` is imported lazily so this module never taints pure import paths.
-- Fail modes apply to NETWORK errors only — contract violations raise before
-  any send (see gate.py) and are never converted to fail-open ALLOWs:
+- Fail modes apply to NETWORK errors only — SDK input contract violations raise
+  before send and malformed successful Core responses raise after receive; neither
+  is converted to a fail-open ALLOW:
     * fail_open (default): return allow-shaped ``EvaluationResult`` with
       ``fallback_used=True`` (callers can tell it apart from a policy ALLOW).
     * fail_closed: raise ``GovernanceAPIError`` (adapters map to native
@@ -200,13 +201,9 @@ async def aevaluate(self, payload: dict) -> EvaluationResult:
         return self._parse_evaluate_response(response)
 
     def _parse_evaluate_response(self, response: Any) -> EvaluationResult:
-        if response.status_code >= 400:
+        if not 200 <= response.status_code < 300:
             return self._network_failure(f"Governance API error: HTTP {response.status_code}")
-        try:
-            data = response.json()
-        except Exception as e:
-            return self._network_failure(f"Governance API returned unparseable body: {e}")
-        result = EvaluationResult.from_dict(data)
+        result = EvaluationResult.from_wire(response.content)
         if result.verdict.should_stop():
             logger.info(f"Governance blocked: {result.reason} (policy: {result.policy_id})")
         return result

openbox_core/contracts/otel_spans.py

@@ -40,6 +40,7 @@ class HookType(str, Enum):
     DB_QUERY = "db_query"
     FILE_OPERATION = "file_operation"
     FUNCTION_CALL = "function_call"
+    SANDBOX_EXECUTION = "sandbox_execution"
     LLM_CALL = "llm_call"  # reserved; disabled until provider hooks are implemented
 
 
@@ -96,6 +97,7 @@ def _attributes_of(obj: Any) -> dict[str, Any]:
     "db_query": "CLIENT",
     "file_operation": "INTERNAL",
     "function_call": "INTERNAL",
+    "sandbox_execution": "INTERNAL",
     "llm_call": "CLIENT",
 }
 
@@ -128,6 +130,9 @@ def _attributes_of(obj: Any) -> dict[str, Any]:
         "bytes_written",
     ),
     "function_call": ("function", "module", "args", "result"),
+    # Sandbox evidence is deliberately attributes-only. It must never gain body,
+    # environment, command, or credential-bearing root fields.
+    "sandbox_execution": (),
 }
 
 

openbox_core/contracts/results.py

@@ -1,20 +1,21 @@
 """Result contracts — Verdict, GuardrailsResult, EvaluationResult, ApprovalResult.
 
 Pure, import-safe module: no network, crypto, OTel, logging, wall-clock, or
-random. Strict dataclass constructors AND loose ``from_dict()`` parsers are
-both public so callers can work with typed values or raw backend dicts.
-
-Parsing preserves ``raw`` so nothing the backend sent is ever lost, and stays
-tolerant of unknown keys (field-shape drift from Core must not crash SDKs).
+random. Successful Core responses use strict ``from_wire()`` parsing;
+``from_dict()`` remains a loose compatibility parser for explicit callers.
+Both preserve unknown keys in ``raw``.
 """
 
 from __future__ import annotations
 
+import json
 import math
 from dataclasses import dataclass, field
 from enum import Enum
 from typing import Any
 
+from ..errors import ContractError
+
 __all__ = [
     "Verdict",
     "GuardrailsResult",
@@ -187,6 +188,77 @@ def _parse_patch(container: dict[str, Any]) -> Patch | None:
     return Patch(new_input=new_input)
 
 
+_WIRE_VERDICTS = {value.value: value for value in Verdict}
+_WIRE_ACTIONS = {
+    **_WIRE_VERDICTS,
+    "continue": Verdict.ALLOW,
+    "stop": Verdict.HALT,
+    "require-approval": Verdict.REQUIRE_APPROVAL,
+    "request_approval": Verdict.REQUIRE_APPROVAL,
+    "request-approval": Verdict.REQUIRE_APPROVAL,
+}
+
+
+def _wire_error() -> ContractError:
+    return ContractError("Malformed governance response", code="RESPONSE_INVALID")
+
+
+def _strict_wire_object(body: bytes) -> dict[str, Any]:
+    if not isinstance(body, bytes):
+        raise _wire_error()
+
+    def unique_object(pairs: list[tuple[str, Any]]) -> dict[str, Any]:
+        result: dict[str, Any] = {}
+        for key, value in pairs:
+            if key in result:
+                raise ValueError
+            result[key] = value
+        return result
+
+    def finite_float(value: str) -> float:
+        result = float(value)
+        if not math.isfinite(result):
+            raise ValueError
+        return result
+
+    def reject_constant(_: str) -> None:
+        raise ValueError
+
+    try:
+        value = json.loads(
+            body.decode("utf-8", errors="strict"),
+            object_pairs_hook=unique_object,
+            parse_constant=reject_constant,
+            parse_float=finite_float,
+        )
+    except (UnicodeDecodeError, json.JSONDecodeError, ValueError, RecursionError):
+        raise _wire_error() from None
+    if not isinstance(value, dict):
+        raise _wire_error()
+    return value
+
+
+def _wire_decision(
+    data: dict[str, Any], field_name: str, vocabulary: dict[str, Verdict]
+) -> Verdict | None:
+    if field_name not in data:
+        return None
+    value = data[field_name]
+    if not isinstance(value, str) or value not in vocabulary:
+        raise _wire_error()
+    return vocabulary[value]
+
+
+def _valid_wire_constraints(value: Any) -> bool:
+    if isinstance(value, dict):
+        return True
+    if not isinstance(value, list):
+        return False
+    return all(isinstance(item, str) for item in value) or all(
+        isinstance(item, dict) for item in value
+    )
+
+
 @dataclass
 class EvaluationResult:
     """Response from a governance evaluation.
@@ -211,7 +283,7 @@ class EvaluationResult:
     trust_tier: str | None = None
     alignment_score: float | None = None
     behavioral_violations: list[str] | None = None
-    constraints: list[dict[str, Any]] | None = None
+    constraints: list[dict[str, Any]] | list[str] | dict[str, Any] | None = None
     fallback_used: bool = False  # True when fail-open produced this result
     diagnostics: list[Any] = field(default_factory=list)
     raw: dict[str, Any] = field(default_factory=dict)
@@ -269,6 +341,32 @@ def from_dict(cls, data: dict[str, Any]) -> EvaluationResult:
             patch=_parse_patch(data),
         )
 
+    @classmethod
+    def from_wire(cls, body: bytes) -> EvaluationResult:
+        """Strictly parse one successful Core evaluation response."""
+        data = _strict_wire_object(body)
+        verdict = _wire_decision(data, "verdict", _WIRE_VERDICTS)
+        action = _wire_decision(data, "action", _WIRE_ACTIONS)
+        if verdict is None and action is None:
+            raise _wire_error()
+        if verdict is not None and action is not None and verdict is not action:
+            raise _wire_error()
+
+        if "fallback_used" in data and type(data["fallback_used"]) is not bool:
+            raise _wire_error()
+        constraints = data.get("constraints")
+        if constraints is not None and not _valid_wire_constraints(constraints):
+            raise _wire_error()
+        for field_name in ("guardrails_result", "guardrails"):
+            if data.get(field_name) is not None and not isinstance(data[field_name], dict):
+                raise _wire_error()
+
+        result = cls.from_dict(data)
+        decision = verdict if verdict is not None else action
+        assert decision is not None
+        result.verdict = decision
+        return result
+
     @classmethod
     def fallback_allow(cls, reason: str) -> EvaluationResult:
         """Allow-shaped result for fail-open network-error paths.

openbox_core/errors.py

@@ -5,7 +5,7 @@
 
 Hierarchy:
     OpenBoxError (base)
-    ├── ContractError               # strict-gate event/runtime contract violation
+    ├── ContractError               # SDK input or Core response contract violation
     ├── OpenBoxConfigError
     │   ├── OpenBoxAuthError
     │   │   └── OpenBoxSigningError # Core rejected a signed (AIP DID) request
@@ -62,11 +62,10 @@ class OpenBoxError(Exception):
 
 
 class ContractError(OpenBoxError):
-    """Raised by the always-strict gate on a malformed event/runtime contract.
+    """Raised when SDK input or a successful Core response violates its contract.
 
-    Contract violations raise *before* any network send, regardless of the
-    ``on_api_error`` fail-open/fail-closed setting — fail-open applies only to
-    network errors, never to contract violations.
+    Input violations raise before network send; response violations raise after
+    a successful exchange. Neither is converted to a fail-open ALLOW result.
 
     Attributes:
         code: Machine-readable violation code (e.g. ``HOOK_TRIGGER_FALSE``).