fix(kernel): fail-closed bound trace structural validation
Return PracticeTraceRefusal before dereferencing bindings or operator results when structural type/shape checks fail. Guard seal_bound zip against binding_result_count_mismatch. Add five refusal-without-exception tests for malformed bound inputs.
This commit is contained in:
parent
484e8ff9fe
commit
b9a6eb9676
2 changed files with 132 additions and 16 deletions
|
|
@ -798,6 +798,22 @@ def _valid_binding_record(binding: object) -> bool:
|
||||||
return all(hasattr(binding, name) for name in required)
|
return all(hasattr(binding, name) for name in required)
|
||||||
|
|
||||||
|
|
||||||
|
_BOUND_STRUCTURAL_REASON_CODES = frozenset(
|
||||||
|
{
|
||||||
|
"invalid_run_type",
|
||||||
|
"invalid_binding_type",
|
||||||
|
"invalid_candidate_operator_result_type",
|
||||||
|
"invalid_replay_result_type",
|
||||||
|
"invalid_replay_refusal_type",
|
||||||
|
"binding_result_count_mismatch",
|
||||||
|
}
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
def _has_bound_structural_failure(reasons: list[str]) -> bool:
|
||||||
|
return any(reason in _BOUND_STRUCTURAL_REASON_CODES for reason in reasons)
|
||||||
|
|
||||||
|
|
||||||
def _valid_operator_result_record(operator_result: object) -> bool:
|
def _valid_operator_result_record(operator_result: object) -> bool:
|
||||||
required = (
|
required = (
|
||||||
"operator_result_id",
|
"operator_result_id",
|
||||||
|
|
@ -1009,9 +1025,17 @@ def build_bound_practice_trace_input(
|
||||||
if not _sha256_text(geometric_search_run_id):
|
if not _sha256_text(geometric_search_run_id):
|
||||||
reasons.append("invalid_geometric_search_run_id")
|
reasons.append("invalid_geometric_search_run_id")
|
||||||
|
|
||||||
|
if isinstance(bindings, tuple) and isinstance(candidate_operator_results, tuple):
|
||||||
if len(bindings) != len(candidate_operator_results):
|
if len(bindings) != len(candidate_operator_results):
|
||||||
reasons.append("binding_result_count_mismatch")
|
reasons.append("binding_result_count_mismatch")
|
||||||
|
|
||||||
|
if _has_bound_structural_failure(reasons):
|
||||||
|
return _refusal(
|
||||||
|
input_digest=None,
|
||||||
|
disposition=_failure_disposition(reasons),
|
||||||
|
reason_codes=tuple(dict.fromkeys(reasons)),
|
||||||
|
)
|
||||||
|
|
||||||
if not reasons and isinstance(run, GeometricSearchRun):
|
if not reasons and isinstance(run, GeometricSearchRun):
|
||||||
reasons.extend(
|
reasons.extend(
|
||||||
_validate_run_bindings(
|
_validate_run_bindings(
|
||||||
|
|
@ -1158,6 +1182,13 @@ def seal_bound_practice_trace(
|
||||||
if replay_refusal_ids != trace_input.replay_refusal_ids:
|
if replay_refusal_ids != trace_input.replay_refusal_ids:
|
||||||
reasons.append("replay_refusal_ids_mismatch")
|
reasons.append("replay_refusal_ids_mismatch")
|
||||||
|
|
||||||
|
if _has_bound_structural_failure(reasons):
|
||||||
|
return _refusal(
|
||||||
|
input_digest=trace_input.input_digest,
|
||||||
|
disposition=_failure_disposition(reasons),
|
||||||
|
reason_codes=tuple(dict.fromkeys(reasons)),
|
||||||
|
)
|
||||||
|
|
||||||
binding_ids = tuple(binding.binding_id for binding in bindings)
|
binding_ids = tuple(binding.binding_id for binding in bindings)
|
||||||
attempt_ids = tuple(binding.candidate_attempt_id for binding in bindings)
|
attempt_ids = tuple(binding.candidate_attempt_id for binding in bindings)
|
||||||
if binding_ids != trace_input.candidate_attempt_binding_ids:
|
if binding_ids != trace_input.candidate_attempt_binding_ids:
|
||||||
|
|
@ -1178,7 +1209,12 @@ def seal_bound_practice_trace(
|
||||||
)
|
)
|
||||||
if len(bindings) != len(candidate_operator_results):
|
if len(bindings) != len(candidate_operator_results):
|
||||||
reasons.append("binding_result_count_mismatch")
|
reasons.append("binding_result_count_mismatch")
|
||||||
for binding, operator_result in zip(bindings, candidate_operator_results, strict=True):
|
elif not reasons:
|
||||||
|
for binding, operator_result in zip(
|
||||||
|
bindings,
|
||||||
|
candidate_operator_results,
|
||||||
|
strict=True,
|
||||||
|
):
|
||||||
reasons.extend(
|
reasons.extend(
|
||||||
_validate_bound_binding_pair(
|
_validate_bound_binding_pair(
|
||||||
run=run,
|
run=run,
|
||||||
|
|
|
||||||
|
|
@ -643,6 +643,86 @@ def test_invalid_schema_versions_refuse() -> None:
|
||||||
assert "invalid_schema_versions" in outcome.reason_codes
|
assert "invalid_schema_versions" in outcome.reason_codes
|
||||||
|
|
||||||
|
|
||||||
|
def test_build_bound_refuses_invalid_binding_type_without_exception() -> None:
|
||||||
|
run, result, _binding = _chain()
|
||||||
|
outcome = build_bound_practice_trace_input(
|
||||||
|
problem_frame_digest=run.problem_frame_digest,
|
||||||
|
original_contract_assessment_id=run.contract_assessment_id,
|
||||||
|
residual_ids=run.residual_ids,
|
||||||
|
search_gate_decision_id=run.gate_decision_id,
|
||||||
|
compute_budget_id=run.budget_id,
|
||||||
|
run=run,
|
||||||
|
bindings=(object(),),
|
||||||
|
candidate_operator_results=(result,),
|
||||||
|
)
|
||||||
|
assert isinstance(outcome, PracticeTraceRefusal)
|
||||||
|
assert "invalid_binding_type" in outcome.reason_codes
|
||||||
|
|
||||||
|
|
||||||
|
def test_build_bound_refuses_invalid_operator_result_type_without_exception() -> None:
|
||||||
|
run, _result, binding = _chain()
|
||||||
|
outcome = build_bound_practice_trace_input(
|
||||||
|
problem_frame_digest=run.problem_frame_digest,
|
||||||
|
original_contract_assessment_id=run.contract_assessment_id,
|
||||||
|
residual_ids=run.residual_ids,
|
||||||
|
search_gate_decision_id=run.gate_decision_id,
|
||||||
|
compute_budget_id=run.budget_id,
|
||||||
|
run=run,
|
||||||
|
bindings=(binding,),
|
||||||
|
candidate_operator_results=(object(),),
|
||||||
|
)
|
||||||
|
assert isinstance(outcome, PracticeTraceRefusal)
|
||||||
|
assert "invalid_candidate_operator_result_type" in outcome.reason_codes
|
||||||
|
|
||||||
|
|
||||||
|
def test_build_bound_refuses_non_tuple_bindings_without_exception() -> None:
|
||||||
|
run, result, _binding = _chain()
|
||||||
|
outcome = build_bound_practice_trace_input(
|
||||||
|
problem_frame_digest=run.problem_frame_digest,
|
||||||
|
original_contract_assessment_id=run.contract_assessment_id,
|
||||||
|
residual_ids=run.residual_ids,
|
||||||
|
search_gate_decision_id=run.gate_decision_id,
|
||||||
|
compute_budget_id=run.budget_id,
|
||||||
|
run=run,
|
||||||
|
bindings=object(), # type: ignore[arg-type]
|
||||||
|
candidate_operator_results=(result,),
|
||||||
|
)
|
||||||
|
assert isinstance(outcome, PracticeTraceRefusal)
|
||||||
|
assert "invalid_binding_type" in outcome.reason_codes
|
||||||
|
|
||||||
|
|
||||||
|
def test_seal_bound_refuses_binding_result_count_mismatch_without_exception() -> None:
|
||||||
|
run, result, binding = _chain()
|
||||||
|
replay = _bound_replay_result(run, result, binding)
|
||||||
|
trace_input = _build_bound_trace_input(run, result, binding, replay_results=(replay,))
|
||||||
|
assert isinstance(trace_input, PracticeTraceInput)
|
||||||
|
outcome = seal_bound_practice_trace(
|
||||||
|
trace_input,
|
||||||
|
run=run,
|
||||||
|
bindings=(binding,),
|
||||||
|
candidate_operator_results=(),
|
||||||
|
replay_results=(replay,),
|
||||||
|
)
|
||||||
|
assert isinstance(outcome, PracticeTraceRefusal)
|
||||||
|
assert "binding_result_count_mismatch" in outcome.reason_codes
|
||||||
|
|
||||||
|
|
||||||
|
def test_seal_bound_refuses_invalid_binding_type_without_exception() -> None:
|
||||||
|
run, result, binding = _chain()
|
||||||
|
replay = _bound_replay_result(run, result, binding)
|
||||||
|
trace_input = _build_bound_trace_input(run, result, binding, replay_results=(replay,))
|
||||||
|
assert isinstance(trace_input, PracticeTraceInput)
|
||||||
|
outcome = seal_bound_practice_trace(
|
||||||
|
trace_input,
|
||||||
|
run=run,
|
||||||
|
bindings=(object(),),
|
||||||
|
candidate_operator_results=(result,),
|
||||||
|
replay_results=(replay,),
|
||||||
|
)
|
||||||
|
assert isinstance(outcome, PracticeTraceRefusal)
|
||||||
|
assert "invalid_binding_type" in outcome.reason_codes
|
||||||
|
|
||||||
|
|
||||||
def test_bound_module_does_not_call_upstream_producers() -> None:
|
def test_bound_module_does_not_call_upstream_producers() -> None:
|
||||||
path = Path("generate/sealed_practice_trace.py")
|
path = Path("generate/sealed_practice_trace.py")
|
||||||
source = path.read_text("utf-8")
|
source = path.read_text("utf-8")
|
||||||
|
|
|
||||||
Loading…
Reference in a new issue