425 lines
19 KiB
Python
425 lines
19 KiB
Python
|
# ============================================================
|
|||
|
|
# P3 偏差⑦⑧ 攻击面专项（Agent-N 验证员补网，全部确定性）：
|
|||
|
|
# 偏差⑦ harness.refresh_confirmation_world_fingerprint 信封哈希同步重算
|
|||
|
|
# —— 攻击面：第二张及以后确认卡的旧信封伪造/篡改/重放必须被拒。
|
|||
|
|
# 偏差⑧ S6 兄弟卡世界指纹重锚 _reanchor_s6_sibling_cards
|
|||
|
|
# —— 攻击面：非兄弟（跨 run / 审批窗口内业务）改动不得借重锚混过漂移检测。
|
|||
|
|
# 手法与 test_fallback_highrisk.py 一致：fake runner / FakeStore / tmp 隔离。
|
|||
|
|
# ============================================================
|
|||
|
|
from __future__ import annotations
|
|||
|
|
|
|||
|
|
import copy
|
|||
|
|
import json
|
|||
|
|
from pathlib import Path
|
|||
|
|
from typing import ClassVar
|
|||
|
|
|
|||
|
|
import pytest
|
|||
|
|
|
|||
|
|
from server.agent_core import fallback_lane, harness
|
|||
|
|
from server.agent_core.providers import reset_provider
|
|||
|
|
from server.aps_domain.workflow import execute_confirmed
|
|||
|
|
from server.contracts import IntentResult
|
|||
|
|
from server.state.checkpoints import CheckpointStore
|
|||
|
|
from server.state.seed import seed_world
|
|||
|
|
|
|||
|
|
DEMO_PRODUCT = "CTRL-A"
|
|||
|
|
|
|||
|
|
|
|||
|
|
@pytest.fixture(autouse=True)
|
|||
|
|
def _isolate(tmp_path, monkeypatch):
|
|||
|
|
monkeypatch.setenv("APS_FALLBACK_DIR", str(tmp_path / "fb"))
|
|||
|
|
monkeypatch.setenv("APS_FEATURES_PATH", str(tmp_path / "features.json"))
|
|||
|
|
monkeypatch.setenv("APS_FALLBACK_HIGHRISK_PATH", str(tmp_path / "fallback-highrisk.json"))
|
|||
|
|
monkeypatch.delenv("LLM_API_KEY", raising=False)
|
|||
|
|
monkeypatch.delenv("LLM_BASE_URL", raising=False)
|
|||
|
|
monkeypatch.delenv("LLM_MODEL", raising=False)
|
|||
|
|
monkeypatch.delenv("LLM_PROVIDER", raising=False)
|
|||
|
|
reset_provider()
|
|||
|
|
from server.state import checkpoints as _cp_mod
|
|||
|
|
from server.state import store as _store_mod
|
|||
|
|
kept_scoped = dict(_store_mod._stores)
|
|||
|
|
kept_cps = dict(_cp_mod._checkpoints)
|
|||
|
|
yield
|
|||
|
|
_store_mod._stores.clear()
|
|||
|
|
_store_mod._stores.update(kept_scoped)
|
|||
|
|
_cp_mod._checkpoints.clear()
|
|||
|
|
_cp_mod._checkpoints.update(kept_cps)
|
|||
|
|
reset_provider()
|
|||
|
|
|
|||
|
|
|
|||
|
|
class FakeStore:
|
|||
|
|
_KIND_TABLE: ClassVar[dict[str, str]] = {
|
|||
|
|
"salesOrder": "salesOrders", "material": "materials",
|
|||
|
|
"audit": "auditEvents", "importBatch": "importBatches"}
|
|||
|
|
|
|||
|
|
def __init__(self, tmp_path: Path):
|
|||
|
|
self.data = seed_world()
|
|||
|
|
self._counters: dict[str, int] = {}
|
|||
|
|
self.tenant_uuid = "platform"
|
|||
|
|
self.world_key = "default"
|
|||
|
|
self.path = str(tmp_path / "world.json")
|
|||
|
|
self.checkpoints = CheckpointStore(str(tmp_path / "checkpoints.json"))
|
|||
|
|
|
|||
|
|
def next_id(self, kind: str) -> int:
|
|||
|
|
if kind not in self._counters:
|
|||
|
|
table = self._KIND_TABLE.get(kind)
|
|||
|
|
self._counters[kind] = max(
|
|||
|
|
(x.get("id", 0) for x in self.data.get(table, [])
|
|||
|
|
if isinstance(x.get("id"), int)), default=0) if table else 0
|
|||
|
|
self._counters[kind] += 1
|
|||
|
|
return self._counters[kind]
|
|||
|
|
|
|||
|
|
def save(self) -> None:
|
|||
|
|
pass
|
|||
|
|
|
|||
|
|
def restore(self, world: dict) -> None:
|
|||
|
|
self.data = copy.deepcopy(world)
|
|||
|
|
self._counters.clear()
|
|||
|
|
|
|||
|
|
|
|||
|
|
def _cfg(tmp_path: Path) -> fallback_lane.FallbackConfig:
|
|||
|
|
return fallback_lane.FallbackConfig(pi_home=str(tmp_path / "pi-home"))
|
|||
|
|
|
|||
|
|
|
|||
|
|
def _write_features(tmp_path: Path) -> None:
|
|||
|
|
(tmp_path / "features.json").write_text(
|
|||
|
|
json.dumps({"version": 1, "features": {"fallback": True}}, ensure_ascii=False),
|
|||
|
|
encoding="utf-8")
|
|||
|
|
|
|||
|
|
|
|||
|
|
def _write_whitelist(tmp_path: Path) -> None:
|
|||
|
|
doc = {"whitelistVersion": 1, "updatedAt": "2026-09-05T10:00:00",
|
|||
|
|
"updatedBy": "attack-test",
|
|||
|
|
"scenarios": {
|
|||
|
|
"S4": {"enabled": True,
|
|||
|
|
"intents": ["flex.simulate_due", "flex.compare",
|
|||
|
|
"scenario.compare", "scenario.sensitivity"],
|
|||
|
|
"roles": ["planner", "admin"]},
|
|||
|
|
"S6": {"enabled": True, "intents": ["mes.report"],
|
|||
|
|
"roles": ["planner", "admin"]},
|
|||
|
|
"S7": {"enabled": True,
|
|||
|
|
"intents": ["agent.fallback.ops.config.apply",
|
|||
|
|
"agent.fallback.policy.update"],
|
|||
|
|
"roles": ["ops", "admin"]}}}
|
|||
|
|
(tmp_path / "fallback-highrisk.json").write_text(
|
|||
|
|
json.dumps(doc, ensure_ascii=False), encoding="utf-8")
|
|||
|
|
|
|||
|
|
|
|||
|
|
def _intent(query: str) -> IntentResult:
|
|||
|
|
return IntentResult(intent="unknown", params={"query": query},
|
|||
|
|
confidence=0.1, source="LLM")
|
|||
|
|
|
|||
|
|
|
|||
|
|
def _fp(world: dict) -> str:
|
|||
|
|
return harness.world_fingerprint(world)
|
|||
|
|
|
|||
|
|
|
|||
|
|
def _pending(confirm_id: str) -> dict:
|
|||
|
|
return harness._approval_store.pending[confirm_id]
|
|||
|
|
|
|||
|
|
|
|||
|
|
def _make_plan_runner(plan_builder):
|
|||
|
|
def runner(task: str, work_dir: Path):
|
|||
|
|
run_dir = work_dir.parent
|
|||
|
|
plan = plan_builder(run_dir)
|
|||
|
|
(run_dir / "outbox").mkdir(parents=True, exist_ok=True)
|
|||
|
|
(run_dir / "outbox" / "plan.json").write_text(
|
|||
|
|
json.dumps(plan, ensure_ascii=False), encoding="utf-8")
|
|||
|
|
yield {"type": "message_end", "message": {"role": "assistant",
|
|||
|
|
"stopReason": "stop", "content": [{"type": "text", "text": "ok"}]}}
|
|||
|
|
yield {"type": "agent_end", "messages": []}
|
|||
|
|
return runner
|
|||
|
|
|
|||
|
|
|
|||
|
|
async def _stage(store: FakeStore, tmp_path: Path, plan: dict, query: str):
|
|||
|
|
reply = await fallback_lane.propose_reply(
|
|||
|
|
store, "s1", _intent(query), runner=_make_plan_runner(lambda rd: plan),
|
|||
|
|
config=_cfg(tmp_path))
|
|||
|
|
confirm_ids = [b.props["confirmId"] for b in (reply.blocks or [])
|
|||
|
|
if getattr(b, "type", "") == "confirm-card"] if reply else []
|
|||
|
|
return reply, confirm_ids
|
|||
|
|
|
|||
|
|
|
|||
|
|
def _approve(store: FakeStore, confirm_id: str, *, approve: bool = True) -> str:
|
|||
|
|
return execute_confirmed(store, confirm_id, approve=approve, actor="attacker-test")
|
|||
|
|
|
|||
|
|
|
|||
|
|
def _seed_s6_world(store: FakeStore, items: list[tuple]) -> None:
|
|||
|
|
store.data.setdefault("workOrders", [])
|
|||
|
|
store.data.setdefault("mesLinks", [])
|
|||
|
|
for wo_id, ext_id, status, pct, qty in items:
|
|||
|
|
store.data["workOrders"].append({
|
|||
|
|
"id": wo_id, "mesExternalId": ext_id, "status": status,
|
|||
|
|
"progressPct": pct, "qtyDone": qty, "productCode": DEMO_PRODUCT})
|
|||
|
|
store.data["mesLinks"].append({
|
|||
|
|
"kind": "dispatch", "woId": wo_id, "externalWoId": ext_id,
|
|||
|
|
"idemKey": f"idem-{wo_id}"})
|
|||
|
|
|
|||
|
|
|
|||
|
|
def _seed_mes_mirror(tmp_path: Path, monkeypatch, count: int):
|
|||
|
|
from server.integrations import mes_stub
|
|||
|
|
from server.integrations.mes_stub import MockMesClient
|
|||
|
|
client = MockMesClient(tmp_path / "mes_mirror.json")
|
|||
|
|
monkeypatch.setattr(mes_stub, "_client", client)
|
|||
|
|
for i in range(count):
|
|||
|
|
client.create_work_order({"productCode": DEMO_PRODUCT}, idem_key=f"idem-m-{i}")
|
|||
|
|
return client
|
|||
|
|
|
|||
|
|
|
|||
|
|
def _s6_plan(items: list[dict]) -> dict:
|
|||
|
|
return {"planVersion": 1, "scenario": "S6", "goal": "MES 断连补录",
|
|||
|
|
"steps": [{"seq": i + 1, "mode": "frozen", "intent": "mes.report",
|
|||
|
|
"summary": f"补录工单 {p['woId']}", "params": p,
|
|||
|
|
"constraints": {}, "expected": []}
|
|||
|
|
for i, p in enumerate(items)]}
|
|||
|
|
|
|||
|
|
|
|||
|
|
def _s3_plan() -> dict:
|
|||
|
|
rows = [{"customerName": "锐扬精密", "productCode": DEMO_PRODUCT,
|
|||
|
|
"quantity": 10, "deliveryDate": "2026-09-20", "orderNo": "RY-9001"}]
|
|||
|
|
return {"planVersion": 1, "scenario": "S3", "goal": "导入订单",
|
|||
|
|
"steps": [{"seq": 1, "mode": "frozen", "intent": "import.commit",
|
|||
|
|
"summary": "导入订单批 1 行",
|
|||
|
|
"params": {"batches": [{"kind": "orders", "rows": rows}]},
|
|||
|
|
"constraints": {"kinds": ["orders"], "maxRows": 500},
|
|||
|
|
"expected": [{"table": "salesOrders", "added": 1}]}]}
|
|||
|
|
|
|||
|
|
|
|||
|
|
def _setup(tmp_path: Path, monkeypatch, wo_ids=(9001, 9002)):
|
|||
|
|
_write_features(tmp_path)
|
|||
|
|
_write_whitelist(tmp_path)
|
|||
|
|
store = FakeStore(tmp_path)
|
|||
|
|
_seed_mes_mirror(tmp_path, monkeypatch, len(wo_ids))
|
|||
|
|
_seed_s6_world(store, [
|
|||
|
|
(w, f"MES-WO-{i + 1:04d}", "RUNNING", 0, 0) for i, w in enumerate(wo_ids)])
|
|||
|
|
return store
|
|||
|
|
|
|||
|
|
|
|||
|
|
# ---------------------------------------------------------------------------
|
|||
|
|
# 偏差⑦ A-1：第二张确认卡信封完整（核心回归——修复前第二张卡 100% 信封自破）
|
|||
|
|
# ---------------------------------------------------------------------------
|
|||
|
|
|
|||
|
|
|
|||
|
|
async def test_a71_second_card_envelope_intact_after_refresh(tmp_path, monkeypatch):
|
|||
|
|
store = _setup(tmp_path, monkeypatch)
|
|||
|
|
items = [{"woId": 9001, "progressPct": 80, "track": "fixed"},
|
|||
|
|
{"woId": 9002, "progressPct": 60, "track": "fixed"}]
|
|||
|
|
_reply, confirm_ids = await _stage(
|
|||
|
|
store, tmp_path, _s6_plan(items), "MES 连不上了，今天的报工帮我补一下")
|
|||
|
|
assert len(confirm_ids) == 2
|
|||
|
|
|
|||
|
|
# 两张卡的信封在出卡指纹推进后都必须通过完整性校验
|
|||
|
|
# （修复前：第二张卡 stage 时捕获到非空 beforeFingerprint，refresh 推进后
|
|||
|
|
# envelopeHash 冻结 → verify_confirmation_envelope 必抛 PermissionError）
|
|||
|
|
for cid in confirm_ids:
|
|||
|
|
harness.verify_confirmation_envelope(_pending(cid)) # 不抛即过
|
|||
|
|
|
|||
|
|
# 先批第二张（乱序审批——S6 逐笔独立语义的直接压力面）
|
|||
|
|
msg2 = _approve(store, confirm_ids[1])
|
|||
|
|
assert "执行完成" in msg2
|
|||
|
|
wos = {w["id"]: w for w in store.data["workOrders"]}
|
|||
|
|
assert wos[9002]["progressPct"] == 60
|
|||
|
|
# 再批第一张（兄弟重锚后信封仍须完整）
|
|||
|
|
harness.verify_confirmation_envelope(_pending(confirm_ids[0]))
|
|||
|
|
msg1 = _approve(store, confirm_ids[0])
|
|||
|
|
assert "执行完成" in msg1
|
|||
|
|
assert wos[9001]["progressPct"] == 80
|
|||
|
|
|
|||
|
|
|
|||
|
|
# ---------------------------------------------------------------------------
|
|||
|
|
# 偏差⑦ A-2：伪造指纹推进（直改 beforeFingerprint 不重算信封）→ 信封拒绝
|
|||
|
|
# ---------------------------------------------------------------------------
|
|||
|
|
|
|||
|
|
|
|||
|
|
async def test_a72_forged_fingerprint_advance_rejected(tmp_path, monkeypatch):
|
|||
|
|
store = _setup(tmp_path, monkeypatch, wo_ids=(9001,))
|
|||
|
|
_reply, confirm_ids = await _stage(
|
|||
|
|
store, tmp_path,
|
|||
|
|
_s6_plan([{"woId": 9001, "progressPct": 80, "track": "fixed"}]),
|
|||
|
|
"MES 连不上了帮我补录")
|
|||
|
|
assert len(confirm_ids) == 1
|
|||
|
|
cid = confirm_ids[0]
|
|||
|
|
fp_before = _fp(store.data)
|
|||
|
|
|
|||
|
|
# 攻击：绕过 refresh_confirmation_world_fingerprint，直接改写冻结指纹并落盘
|
|||
|
|
# （等价于直接篡改 approvals.json——decide 每次事务从磁盘刷新，纯内存篡改
|
|||
|
|
# 会被刷新抹掉，本身就是一层防御；这里按真实攻击面改盘）
|
|||
|
|
_pending(cid)["beforeFingerprint"] = "0" * 64
|
|||
|
|
harness._approval_store.save()
|
|||
|
|
with pytest.raises(PermissionError, match="完整性校验失败"):
|
|||
|
|
harness.verify_confirmation_envelope(_pending(cid))
|
|||
|
|
message = _approve(store, cid)
|
|||
|
|
assert "确认信封证据校验未通过" in message
|
|||
|
|
assert _fp(store.data) == fp_before # 零写入
|
|||
|
|
wos = {w["id"]: w for w in store.data["workOrders"]}
|
|||
|
|
assert wos[9001]["progressPct"] == 0
|
|||
|
|
|
|||
|
|
|
|||
|
|
# ---------------------------------------------------------------------------
|
|||
|
|
# 偏差⑦ A-3：篡改冻结参数（progressPct 80→100）→ 参数摘要拒绝
|
|||
|
|
# ---------------------------------------------------------------------------
|
|||
|
|
|
|||
|
|
|
|||
|
|
async def test_a73_tampered_params_rejected(tmp_path, monkeypatch):
|
|||
|
|
store = _setup(tmp_path, monkeypatch, wo_ids=(9001,))
|
|||
|
|
_reply, confirm_ids = await _stage(
|
|||
|
|
store, tmp_path,
|
|||
|
|
_s6_plan([{"woId": 9001, "progressPct": 80, "track": "fixed"}]),
|
|||
|
|
"MES 连不上了帮我补录")
|
|||
|
|
cid = confirm_ids[0]
|
|||
|
|
fp_before = _fp(store.data)
|
|||
|
|
|
|||
|
|
_pending(cid)["params"]["plan"]["steps"][0]["params"]["progressPct"] = 100
|
|||
|
|
harness._approval_store.save() # 篡改落盘（真实攻击面）
|
|||
|
|
with pytest.raises(PermissionError, match="参数摘要不一致"):
|
|||
|
|
harness.verify_confirmation_envelope(_pending(cid))
|
|||
|
|
message = _approve(store, cid)
|
|||
|
|
assert "确认信封证据校验未通过" in message
|
|||
|
|
assert _fp(store.data) == fp_before
|
|||
|
|
wos = {w["id"]: w for w in store.data["workOrders"]}
|
|||
|
|
assert wos[9001]["progressPct"] == 0 # 未按篡改值落账
|
|||
|
|
|
|||
|
|
|
|||
|
|
# ---------------------------------------------------------------------------
|
|||
|
|
# 偏差⑦ A-4：重放已执行的确认卡 → 明确失效，不重复落账
|
|||
|
|
# ---------------------------------------------------------------------------
|
|||
|
|
|
|||
|
|
|
|||
|
|
async def test_a74_replay_executed_card_rejected(tmp_path, monkeypatch):
|
|||
|
|
store = _setup(tmp_path, monkeypatch, wo_ids=(9001,))
|
|||
|
|
_reply, confirm_ids = await _stage(
|
|||
|
|
store, tmp_path,
|
|||
|
|
_s6_plan([{"woId": 9001, "progressPct": 80, "track": "fixed"}]),
|
|||
|
|
"MES 连不上了帮我补录")
|
|||
|
|
cid = confirm_ids[0]
|
|||
|
|
msg = _approve(store, cid)
|
|||
|
|
assert "执行完成" in msg
|
|||
|
|
wos = {w["id"]: w for w in store.data["workOrders"]}
|
|||
|
|
assert wos[9001]["progressPct"] == 80
|
|||
|
|
|
|||
|
|
replay = _approve(store, cid) # 同 confirmId 重放
|
|||
|
|
assert "已失效" in replay
|
|||
|
|
assert wos[9001]["progressPct"] == 80 # 未重复落账
|
|||
|
|
exec_audits = [e for e in store.data.get("auditEvents", [])
|
|||
|
|
if e.get("action") == "agent.fallback.execute"
|
|||
|
|
and e.get("category") == "WORLD_WRITE"
|
|||
|
|
and e.get("result") == "SUCCESS"]
|
|||
|
|
assert len(exec_audits) == 1 # 只有首次执行的成功总账
|
|||
|
|
|
|||
|
|
|
|||
|
|
# ---------------------------------------------------------------------------
|
|||
|
|
# 偏差⑧ B-1：重锚只认同 run 兄弟卡——跨 run 卡指纹不推进，漂移检测照常拦
|
|||
|
|
# ---------------------------------------------------------------------------
|
|||
|
|
|
|||
|
|
|
|||
|
|
async def test_b81_reanchor_only_touches_same_run_siblings(tmp_path, monkeypatch):
|
|||
|
|
store = _setup(tmp_path, monkeypatch, wo_ids=(9001, 9002, 9003))
|
|||
|
|
# run A：两笔补录
|
|||
|
|
_reply_a, cards_a = await _stage(
|
|||
|
|
store, tmp_path,
|
|||
|
|
_s6_plan([{"woId": 9001, "progressPct": 80, "track": "fixed"},
|
|||
|
|
{"woId": 9002, "progressPct": 60, "track": "fixed"}]),
|
|||
|
|
"MES 连不上了，今天的报工帮我补一下")
|
|||
|
|
# run B：另一轮补录（跨 run 非兄弟）
|
|||
|
|
_reply_b, cards_b = await _stage(
|
|||
|
|
store, tmp_path,
|
|||
|
|
_s6_plan([{"woId": 9003, "progressPct": 50, "track": "fixed"}]),
|
|||
|
|
"MES 连不上了，今天的报工帮我补一下")
|
|||
|
|
assert len(cards_a) == 2 and len(cards_b) == 1
|
|||
|
|
run_a = _pending(cards_a[0])["params"]["runId"]
|
|||
|
|
run_b = _pending(cards_b[0])["params"]["runId"]
|
|||
|
|
assert run_a != run_b
|
|||
|
|
fp_b_frozen = _pending(cards_b[0])["beforeFingerprint"]
|
|||
|
|
|
|||
|
|
msg_a0 = _approve(store, cards_a[0]) # 执行 A 第一笔
|
|||
|
|
assert "执行完成" in msg_a0
|
|||
|
|
|
|||
|
|
# 兄弟卡 A2 被重锚：指纹推进到当前世界且信封完整 → 可正常批准
|
|||
|
|
assert _pending(cards_a[1])["beforeFingerprint"] == _fp(store.data)
|
|||
|
|
harness.verify_confirmation_envelope(_pending(cards_a[1]))
|
|||
|
|
msg_a1 = _approve(store, cards_a[1])
|
|||
|
|
assert "执行完成" in msg_a1
|
|||
|
|
|
|||
|
|
# 跨 run 卡 B 未被重锚（指纹停留原值）→ 漂移检测拦截，fail-closed
|
|||
|
|
assert _pending(cards_b[0])["beforeFingerprint"] == fp_b_frozen
|
|||
|
|
assert _pending(cards_b[0])["beforeFingerprint"] != _fp(store.data)
|
|||
|
|
msg_b = _approve(store, cards_b[0])
|
|||
|
|
assert "世界指纹漂移" in msg_b
|
|||
|
|
wos = {w["id"]: w for w in store.data["workOrders"]}
|
|||
|
|
assert wos[9003]["progressPct"] == 0 # B 未落账
|
|||
|
|
assert wos[9001]["progressPct"] == 80 # A 已落账不受影响
|
|||
|
|
denied = [e for e in store.data.get("auditEvents", [])
|
|||
|
|
if e.get("action") == "agent.fallback.execute"
|
|||
|
|
and e.get("category") == "WORLD_WRITE"
|
|||
|
|
and e.get("result") == "DENIED"]
|
|||
|
|
assert denied # 漂移拒绝有 DENIED 审计
|
|||
|
|
|
|||
|
|
|
|||
|
|
# ---------------------------------------------------------------------------
|
|||
|
|
# 偏差⑧ B-2（核心攻击）：重锚后审批窗口内的非兄弟业务改动仍被漂移检测拦下
|
|||
|
|
# ---------------------------------------------------------------------------
|
|||
|
|
|
|||
|
|
|
|||
|
|
async def test_b82_non_sibling_business_change_still_blocked(tmp_path, monkeypatch):
|
|||
|
|
store = _setup(tmp_path, monkeypatch)
|
|||
|
|
items = [{"woId": 9001, "progressPct": 80, "track": "fixed"},
|
|||
|
|
{"woId": 9002, "progressPct": 60, "track": "fixed"}]
|
|||
|
|
_reply, confirm_ids = await _stage(
|
|||
|
|
store, tmp_path, _s6_plan(items), "MES 连不上了，今天的报工帮我补一下")
|
|||
|
|
assert len(confirm_ids) == 2
|
|||
|
|
|
|||
|
|
msg1 = _approve(store, confirm_ids[0]) # 第一笔落账 → 兄弟重锚
|
|||
|
|
assert "执行完成" in msg1
|
|||
|
|
assert _pending(confirm_ids[1])["beforeFingerprint"] == _fp(store.data)
|
|||
|
|
|
|||
|
|
# 攻击/并发事实：重锚之后、第二笔批准之前，发生非兄弟业务改动
|
|||
|
|
# （另一用户的正常业务写——若借重锚混过检测，第二笔将按过时口径落账）
|
|||
|
|
store.data.setdefault("salesOrders", []).append(
|
|||
|
|
{"id": 777001, "orderNo": "ATTACK-WINDOW", "productCode": DEMO_PRODUCT,
|
|||
|
|
"quantity": 1, "deliveryDate": "2026-09-25"})
|
|||
|
|
store.save()
|
|||
|
|
|
|||
|
|
msg2 = _approve(store, confirm_ids[1])
|
|||
|
|
assert "世界指纹漂移" in msg2 # 被拦下：重锚没有放行非兄弟改动
|
|||
|
|
wos = {w["id"]: w for w in store.data["workOrders"]}
|
|||
|
|
assert wos[9002]["progressPct"] == 0 # 第二笔未落账
|
|||
|
|
assert wos[9001]["progressPct"] == 80 # 第一笔已落账不动
|
|||
|
|
|
|||
|
|
|
|||
|
|
# ---------------------------------------------------------------------------
|
|||
|
|
# 偏差⑧ B-3：伪造兄弟（篡改 runId 蹭重锚）→ 参数摘要破损，信封层拒绝
|
|||
|
|
# ---------------------------------------------------------------------------
|
|||
|
|
|
|||
|
|
|
|||
|
|
async def test_b83_forged_sibling_runid_rejected(tmp_path, monkeypatch):
|
|||
|
|
store = _setup(tmp_path, monkeypatch)
|
|||
|
|
_reply_a, cards_a = await _stage(
|
|||
|
|
store, tmp_path,
|
|||
|
|
_s6_plan([{"woId": 9001, "progressPct": 80, "track": "fixed"},
|
|||
|
|
{"woId": 9002, "progressPct": 60, "track": "fixed"}]),
|
|||
|
|
"MES 连不上了，今天的报工帮我补一下")
|
|||
|
|
_reply_b, cards_b = await _stage(
|
|||
|
|
store, tmp_path, _s3_plan(), "把这份客户表格导进来")
|
|||
|
|
assert len(cards_a) == 2 and len(cards_b) == 1
|
|||
|
|
run_a = _pending(cards_a[0])["params"]["runId"]
|
|||
|
|
orders_before = len(store.data.get("salesOrders") or [])
|
|||
|
|
|
|||
|
|
# 攻击：把非兄弟卡（S3 导入卡）的 runId 篡改为 S6 run 并落盘，试图让重锚
|
|||
|
|
# 免费推进它的冻结指纹、混过漂移检测
|
|||
|
|
_pending(cards_b[0])["params"]["runId"] = run_a
|
|||
|
|
harness._approval_store.save()
|
|||
|
|
msg_a0 = _approve(store, cards_a[0]) # 触发重锚
|
|||
|
|
assert "执行完成" in msg_a0
|
|||
|
|
# 重锚按匹配条件确实推进了被篡改卡（机制如实运作）……
|
|||
|
|
assert _pending(cards_b[0])["beforeFingerprint"] == _fp(store.data)
|
|||
|
|
# ……但 params 被篡改 → paramsHash 破损，信封层物理拒绝
|
|||
|
|
msg_b = _approve(store, cards_b[0])
|
|||
|
|
assert "确认信封证据校验未通过" in msg_b
|
|||
|
|
assert len(store.data.get("salesOrders") or []) == orders_before # 导入未执行
|
|||
|
|
|
|||
|
|
# 真兄弟卡不受伪造事件影响，仍可正常批准落账
|
|||
|
|
msg_a1 = _approve(store, cards_a[1])
|
|||
|
|
assert "执行完成" in msg_a1
|
|||
|
|
wos = {w["id"]: w for w in store.data["workOrders"]}
|
|||
|
|
assert wos[9002]["progressPct"] == 60
|