300 lines
13 KiB
Python
300 lines
13 KiB
Python
#!/usr/bin/env python3
|
||
"""GateInputBuilder 的安全输入与同 ID 混淆项测试。"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import copy
|
||
import pathlib
|
||
import sys
|
||
import unittest
|
||
|
||
PROJECT_ROOT = pathlib.Path(__file__).resolve().parents[3]
|
||
SKILLS_DIR = PROJECT_ROOT / ".agent" / "skills"
|
||
SCRIPT_DIR = PROJECT_ROOT / "muse" / "lifecycle" / "quality" / "skills" / "mechanical" / "adjudicate-quality-gate" / "scripts"
|
||
TEST_DIR = PROJECT_ROOT / "tests" / "skills" / "adjudicate-quality-gate"
|
||
QUALITY_GATE_DIR = PROJECT_ROOT / "muse" / "lifecycle" / "quality" / "skills" / "judge" / "score-content-quality" / "scripts"
|
||
for _import_dir in (SCRIPT_DIR, TEST_DIR, QUALITY_GATE_DIR):
|
||
if str(_import_dir) not in sys.path:
|
||
sys.path.insert(0, str(_import_dir))
|
||
|
||
from gate_input_builder import GateInputBuildError, GateInputBuilder, canonical_sha256
|
||
from test_writer_gate import build_input, source_bundle
|
||
from writer_rubric import RUBRIC_POLICY_VERSION
|
||
|
||
|
||
class GateInputBuilderTest(unittest.TestCase):
|
||
@staticmethod
|
||
def _rehash_wrapper(wrapper):
|
||
"""只重签 wrapper，故意不替攻击者重签生产内层回执。"""
|
||
|
||
wrapper["receiptSha256"] = canonical_sha256(
|
||
{key: value for key, value in wrapper.items() if key != "receiptSha256"}
|
||
)
|
||
|
||
def test_builder_emits_v3_policy_binding_and_complete_confounders(self):
|
||
gate_input = build_input(source_bundle())
|
||
self.assertEqual(gate_input["schemaVersion"], "writer-gate-input-v3")
|
||
self.assertEqual(gate_input["rubricPolicyVersion"], RUBRIC_POLICY_VERSION)
|
||
self.assertTrue(gate_input["builderReceiptSha256"].startswith("sha256:"))
|
||
self.assertEqual(
|
||
set(gate_input["samples"][0]["confounders"]),
|
||
{
|
||
"falseNegative",
|
||
"falsePositive",
|
||
"leakage",
|
||
"reviewerInstability",
|
||
"newCharactersWithoutCards",
|
||
},
|
||
)
|
||
|
||
def test_sensitive_manifest_field_is_rejected(self):
|
||
bundle = source_bundle()
|
||
bundle["manifest"]["samples"][0]["candidateBody"] = "正文不得进入安全 manifest"
|
||
with self.assertRaisesRegex(GateInputBuildError, "敏感字段"):
|
||
GateInputBuilder().build(**bundle)
|
||
|
||
def test_same_id_and_candidate_binding_are_mandatory(self):
|
||
bundle = source_bundle()
|
||
verdict = bundle["detector_reports"]["sample-1"]["A"]["assertionVerdicts"][0]
|
||
verdict["assertionId"] = "forged-id"
|
||
report = bundle["detector_reports"]["sample-1"]["A"]
|
||
from gate_input_builder import canonical_sha256
|
||
report["reportSha256"] = canonical_sha256(
|
||
{key: value for key, value in report.items() if key != "reportSha256"}
|
||
)
|
||
gate_input = build_input(bundle)
|
||
self.assertTrue(gate_input["samples"][0]["systemFailure"])
|
||
self.assertFalse(gate_input["samples"][0]["schemaValid"])
|
||
|
||
def test_detector_unknown_is_quality_signal_and_reduces_c_coverage(self):
|
||
"""合法 needs_evidence 可进入 Gate；unknown 不计入 C 臂硬约束覆盖率。"""
|
||
|
||
bundle = source_bundle(1, gate="A")
|
||
report = bundle["detector_reports"]["sample-1"]["C"]
|
||
report["status"] = "needs_evidence"
|
||
report["hardConstraintVerdicts"][0]["verdict"] = "unknown"
|
||
report["reportSha256"] = canonical_sha256(
|
||
{key: value for key, value in report.items() if key != "reportSha256"}
|
||
)
|
||
|
||
gate_input = build_input(bundle)
|
||
|
||
sample = gate_input["samples"][0]
|
||
self.assertTrue(sample["schemaValid"], sample)
|
||
self.assertFalse(sample["systemFailure"], sample)
|
||
self.assertEqual(sample["cArm"]["hardConstraintCoverage"], 0.0)
|
||
|
||
def test_empty_wrong_role_and_unbound_inner_execution_receipts_fail_closed(self):
|
||
"""wrapper 自哈希不能替代生产 runtime 原始 ExecutionReceipt 绑定。"""
|
||
|
||
mutations = {
|
||
"empty": lambda wrapper: wrapper.update(
|
||
{"executionReceipts": [], "executionReceiptSha256": []}
|
||
),
|
||
"wrong_role": lambda wrapper: wrapper["executionReceipts"][0].update(
|
||
{"adapterRole": "semantic_detector"}
|
||
),
|
||
"unbound": lambda wrapper: wrapper["executionReceipts"][0].update(
|
||
{"inputSha256": "sha256:" + "f" * 64}
|
||
),
|
||
}
|
||
expected_reason = {
|
||
"empty": "A_writer_inner_receipts_empty",
|
||
"wrong_role": "A_writer_inner_receipt_role_mismatch",
|
||
"unbound": "A_writer_inner_receipt_hash_mismatch",
|
||
}
|
||
for name, mutate in mutations.items():
|
||
bundle = source_bundle(1, gate="A")
|
||
wrapper = bundle["execution_receipts"]["sample-1"]["A"]["writer"]
|
||
mutate(wrapper)
|
||
self._rehash_wrapper(wrapper)
|
||
|
||
gate_input = build_input(bundle)
|
||
|
||
sample = gate_input["samples"][0]
|
||
with self.subTest(name=name):
|
||
self.assertTrue(sample["systemFailure"])
|
||
self.assertIn(expected_reason[name], sample["systemFailureReasons"])
|
||
|
||
def test_writer_and_detector_retry_chains_bind_to_final_receipt(self):
|
||
"""1..3 条重试回执均保留，semantic 报告必须绑定最后一条。"""
|
||
|
||
bundle = source_bundle(1, gate="A")
|
||
for arm in ("A", "B", "C"):
|
||
for role in ("writer", "semantic_detector"):
|
||
wrapper = bundle["execution_receipts"]["sample-1"][arm][role]
|
||
retry = copy.deepcopy(wrapper["executionReceipts"][0])
|
||
retry["invocationId"] = f"{role}-{arm}-retry-2"
|
||
wrapper["executionReceipts"].append(retry)
|
||
wrapper["executionReceiptSha256"].append(canonical_sha256(retry))
|
||
self._rehash_wrapper(wrapper)
|
||
detector = bundle["detector_reports"]["sample-1"][arm]
|
||
detector["modelReceiptSha256"] = bundle["execution_receipts"]["sample-1"][arm][
|
||
"semantic_detector"
|
||
]["executionReceiptSha256"][-1]
|
||
detector["reportSha256"] = canonical_sha256(
|
||
{key: value for key, value in detector.items() if key != "reportSha256"}
|
||
)
|
||
|
||
gate_input = build_input(bundle)
|
||
|
||
self.assertFalse(gate_input["samples"][0]["systemFailure"], gate_input["samples"][0])
|
||
|
||
def test_nonfinal_known_cost_api_error_receipt_can_precede_detector_success(self):
|
||
"""前置 API 失败必须保留审计；只有末笔成功且报告绑定末笔时可进入 Gate。"""
|
||
|
||
bundle = source_bundle(1, gate="A")
|
||
wrapper = bundle["execution_receipts"]["sample-1"]["A"]["semantic_detector"]
|
||
transient = copy.deepcopy(wrapper["executionReceipts"][0])
|
||
transient.update(
|
||
{
|
||
"invocationId": "semantic-A-transient-api-error",
|
||
"actualModelId": None,
|
||
"modelMatch": False,
|
||
"totalCostUsd": "0.000000",
|
||
"terminalReason": "api_error",
|
||
"isError": True,
|
||
"apiErrorStatus": None,
|
||
"exitCode": 1,
|
||
"structuredOutputSha256": None,
|
||
}
|
||
)
|
||
wrapper["executionReceipts"].insert(0, transient)
|
||
wrapper["executionReceiptSha256"].insert(0, canonical_sha256(transient))
|
||
self._rehash_wrapper(wrapper)
|
||
|
||
gate_input = build_input(bundle)
|
||
|
||
self.assertFalse(gate_input["samples"][0]["systemFailure"], gate_input["samples"][0])
|
||
|
||
def test_final_api_error_receipt_remains_system_failure(self):
|
||
"""API 错误只能出现在最终成功前；把它放在末笔不得借重试合同放行。"""
|
||
|
||
bundle = source_bundle(1, gate="A")
|
||
wrapper = bundle["execution_receipts"]["sample-1"]["A"]["semantic_detector"]
|
||
final_error = copy.deepcopy(wrapper["executionReceipts"][0])
|
||
final_error.update(
|
||
{
|
||
"actualModelId": None,
|
||
"modelMatch": False,
|
||
"totalCostUsd": "0.000000",
|
||
"terminalReason": "api_error",
|
||
"isError": True,
|
||
"apiErrorStatus": None,
|
||
"exitCode": 1,
|
||
"structuredOutputSha256": None,
|
||
}
|
||
)
|
||
wrapper["executionReceipts"] = [final_error]
|
||
wrapper["executionReceiptSha256"] = [canonical_sha256(final_error)]
|
||
self._rehash_wrapper(wrapper)
|
||
|
||
gate_input = build_input(bundle)
|
||
|
||
sample = gate_input["samples"][0]
|
||
self.assertTrue(sample["systemFailure"])
|
||
self.assertIn(
|
||
"A_semantic_detector_inner_receipt_not_successful",
|
||
sample["systemFailureReasons"],
|
||
)
|
||
|
||
def test_four_writer_receipts_fail_closed(self):
|
||
"""有界修订最多三次，第四条回执不得借审计链绕过。"""
|
||
|
||
bundle = source_bundle(1, gate="A")
|
||
wrapper = bundle["execution_receipts"]["sample-1"]["A"]["writer"]
|
||
for invocation in range(2, 6):
|
||
retry = copy.deepcopy(wrapper["executionReceipts"][0])
|
||
retry["invocationId"] = f"writer-A-retry-{invocation}"
|
||
wrapper["executionReceipts"].append(retry)
|
||
wrapper["executionReceiptSha256"].append(canonical_sha256(retry))
|
||
self._rehash_wrapper(wrapper)
|
||
|
||
gate_input = build_input(bundle)
|
||
|
||
self.assertTrue(gate_input["samples"][0]["systemFailure"])
|
||
self.assertIn(
|
||
"A_writer_inner_receipt_count_invalid",
|
||
gate_input["samples"][0]["systemFailureReasons"],
|
||
)
|
||
|
||
def test_judge_score_tampering_without_original_hash_resign_fails_closed(self):
|
||
"""篡改稳定报告分数但保留生产原始 hash 时不得进入 Gate 分数恢复。"""
|
||
|
||
bundle = source_bundle(1, gate="A")
|
||
report = bundle["judge_reports"]["sample-1"]
|
||
original_report_hash = report["reportSha256"]
|
||
report["candidateScores"][0]["scores"]["setting_entity_fidelity"] = 999
|
||
|
||
gate_input = build_input(bundle)
|
||
|
||
sample = gate_input["samples"][0]
|
||
self.assertEqual(report["reportSha256"], original_report_hash)
|
||
self.assertTrue(sample["systemFailure"])
|
||
self.assertFalse(sample["schemaValid"])
|
||
self.assertTrue(
|
||
any("judgeReports.sample-1.reportSha256 不一致" in reason for reason in sample["systemFailureReasons"])
|
||
)
|
||
|
||
def test_judge_report_resign_cannot_replace_original_panel_receipt_binding(self):
|
||
"""攻击者重签报告自哈希后，仍不能替换生产 panel 的原始模型回执集合。"""
|
||
|
||
bundle = source_bundle(1, gate="A")
|
||
report = bundle["judge_reports"]["sample-1"]
|
||
report["modelReceiptSha256"] = "sha256:" + "e" * 64
|
||
report["reportSha256"] = canonical_sha256(
|
||
{key: value for key, value in report.items() if key != "reportSha256"}
|
||
)
|
||
|
||
gate_input = build_input(bundle)
|
||
|
||
sample = gate_input["samples"][0]
|
||
self.assertTrue(sample["systemFailure"])
|
||
self.assertFalse(sample["schemaValid"])
|
||
self.assertTrue(
|
||
any("未绑定原始 panel 回执" in reason for reason in sample["systemFailureReasons"])
|
||
)
|
||
|
||
def test_judge_report_cannot_swap_preregistered_scenario_policy(self):
|
||
bundle = source_bundle(1, gate="A")
|
||
panel = bundle["judge_reports"]["sample-1"]
|
||
panel["reviewerReports"][0]["scenario"] = "character_dialogue"
|
||
panel["reportSha256"] = canonical_sha256(
|
||
{key: value for key, value in panel.items() if key != "reportSha256"}
|
||
)
|
||
|
||
gate_input = build_input(bundle)
|
||
|
||
sample = gate_input["samples"][0]
|
||
self.assertTrue(sample["systemFailure"])
|
||
self.assertFalse(sample["schemaValid"])
|
||
self.assertTrue(
|
||
any(
|
||
"未绑定当前样本场景评分策略" in reason
|
||
for reason in sample["systemFailureReasons"]
|
||
)
|
||
)
|
||
|
||
def test_legal_score_tampering_and_report_resign_still_fails_receipt_binding(self):
|
||
"""合法 9.5 分即使重签 panel 自哈希，也必须与模型原始 structured output 冲突。"""
|
||
|
||
bundle = source_bundle(1, gate="A")
|
||
report = bundle["judge_reports"]["sample-1"]
|
||
report["candidateScores"][0]["scores"]["setting_entity_fidelity"] = 9.5
|
||
report["reportSha256"] = canonical_sha256(
|
||
{key: value for key, value in report.items() if key != "reportSha256"}
|
||
)
|
||
|
||
gate_input = build_input(bundle)
|
||
|
||
sample = gate_input["samples"][0]
|
||
self.assertTrue(sample["systemFailure"])
|
||
self.assertFalse(sample["schemaValid"])
|
||
self.assertTrue(
|
||
any("模型原始 structured output" in reason for reason in sample["systemFailureReasons"])
|
||
)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
unittest.main()
|