From 3b6451ba4f2c388b7f9f62176446b7247369192b Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Mon, 31 Aug 2026 03:32:36 +0800 Subject: [PATCH] test: add reconciliation recovery fault injection Co-Authored-By: Codex --- ...econciliation_baseline_enrollment.zh-CN.md | 8 ++ ...reconciliation_recovery_fault_injection.py | 108 ++++++++++++++++++ 2 files changed, 116 insertions(+) create mode 100644 tests/test_reconciliation_recovery_fault_injection.py diff --git a/docs/ibkr_reconciliation_baseline_enrollment.zh-CN.md b/docs/ibkr_reconciliation_baseline_enrollment.zh-CN.md index 54d119d..4d05c87 100644 --- a/docs/ibkr_reconciliation_baseline_enrollment.zh-CN.md +++ b/docs/ibkr_reconciliation_baseline_enrollment.zh-CN.md @@ -45,3 +45,11 @@ AIAuditBridge 的完整 `reconciliation_baseline` 输出。它会同时核验： `execution_authority_granted=false`、`state_write_attempted=false`；它没有 Cloud Run、 券商、执行标记或订单写入代码。下一层单独的最小权限控制器才可消费该计划，并且仍要 在同一目标上比较五项摘要后执行一次精确 CAS。 + +## 故障注入回归 + +恢复链路的回归测试会主动注入：控制台把不可执行策略篡改为可执行、五项摘要之一 +漂移、人工确认过期、确认与新回执同秒、运行目标不再是 `RECONCILE_ONLY`、以及控制器 +令牌被试图发送到普通管理站路径。每一种情况都必须抛出拒绝或返回没有 +`transition_plan` 的结果；测试同时断言 `state_write_attempted=false`。这让后续接入 +最小权限 CAS 时能持续证明“异常只能保持冻结，不能意外恢复实盘”。 diff --git a/tests/test_reconciliation_recovery_fault_injection.py b/tests/test_reconciliation_recovery_fault_injection.py new file mode 100644 index 0000000..908dc45 --- /dev/null +++ b/tests/test_reconciliation_recovery_fault_injection.py @@ -0,0 +1,108 @@ +"""Fault injection for the non-executable IBKR legacy recovery path. + +Each case simulates an untrusted or stale input at a trust boundary. A test +passes only when the verifier rejects it or returns no transition plan; none of +these paths has a state-write or broker-order implementation. +""" + +from __future__ import annotations + +from datetime import datetime, timedelta, timezone + +import pytest + +from quant_platform_kit.common.broker_reconciliation import ( + calculate_broker_reconciliation_evidence_sha256, +) + +from tests.test_reconciliation_recovery_verify_only import ( + _candidate_payload, + _console_response, + _dual_review, + _evidence, + _runtime_target, +) +from scripts.verify_reconciliation_recovery import ( + read_console_confirmation, + verify_reconciliation_recovery, +) + + +def _inputs(start: datetime) -> tuple[dict[str, object], str]: + candidate = _candidate_payload(start) + candidate_value = candidate["candidate"] + assert isinstance(candidate_value, dict) + return candidate, str(candidate_value["candidate_sha256"]) + + +def test_fault_injection_tampered_console_policy_is_rejected_before_evidence_evaluation() -> None: + start = datetime(2026, 8, 31, 1, 0, tzinfo=timezone.utc) + candidate, candidate_sha256 = _inputs(start) + confirmation = _console_response(candidate_sha256, confirmed_at=start + timedelta(minutes=3)) + confirmation["policy"] = { + "no_order": True, + "execution_authority_granted": True, + "controller_must_reverify": True, + } + + with pytest.raises(ValueError, match="non-execution policy"): + verify_reconciliation_recovery( + candidate_payload=candidate, + dual_review_payload=_dual_review(candidate_sha256), + confirmation_payload=confirmation, + current_receipt_payload={"evidence": _evidence(observed_at=start + timedelta(minutes=4)).to_dict()}, + runtime_target_payload=_runtime_target(), + recovery_id="ibkr-soxl-live-recovery", + now=start + timedelta(minutes=5), + ) + + +def test_fault_injection_current_cash_digest_drift_closes_transition_plan() -> None: + start = datetime(2026, 8, 31, 1, 0, tzinfo=timezone.utc) + candidate, candidate_sha256 = _inputs(start) + current = _evidence(observed_at=start + timedelta(minutes=4)).to_dict() + current["cash_sha256"] = "9" * 64 + current["evidence_sha256"] = calculate_broker_reconciliation_evidence_sha256(current) + + result = verify_reconciliation_recovery( + candidate_payload=candidate, + dual_review_payload=_dual_review(candidate_sha256), + confirmation_payload=_console_response(candidate_sha256, confirmed_at=start + timedelta(minutes=3)), + current_receipt_payload={"evidence": current}, + runtime_target_payload=_runtime_target(), + recovery_id="ibkr-soxl-live-recovery", + now=start + timedelta(minutes=5), + ) + + assert result["ready_for_atomic_state_transition"] is False + assert result["transition_plan"] is None + assert "broker_reconciliation_cash_mismatch" in result["findings"] + assert result["policy"]["state_write_attempted"] is False # type: ignore[index] + + +def test_fault_injection_stale_human_confirmation_closes_transition_plan() -> None: + start = datetime(2026, 8, 31, 1, 0, tzinfo=timezone.utc) + candidate, candidate_sha256 = _inputs(start) + + result = verify_reconciliation_recovery( + candidate_payload=candidate, + dual_review_payload=_dual_review(candidate_sha256), + confirmation_payload=_console_response(candidate_sha256, confirmed_at=start + timedelta(minutes=3)), + current_receipt_payload={"evidence": _evidence(observed_at=start + timedelta(minutes=40)).to_dict()}, + runtime_target_payload=_runtime_target(), + recovery_id="ibkr-soxl-live-recovery", + now=start + timedelta(minutes=40), + ) + + assert result["ready_for_atomic_state_transition"] is False + assert result["transition_plan"] is None + assert "reconciliation_recovery_confirmation_stale" in result["findings"] + + +def test_fault_injection_controller_read_url_cannot_be_a_general_console_endpoint() -> None: + with pytest.raises(ValueError, match="controller read endpoint"): + read_console_confirmation( + confirmation_url="https://console.example/api/reconciliation-recovery", + recovery_id="ibkr-soxl-live-recovery", + token="controller-token", + )