code-to10 v6 A3: enforce signer separation on audit-verdict submission

process_audit_verdict now calls validate_signer_separation
before any state save or ledger write (previously the function
had no production caller on this path: it only ran inside
collect_failed_checks via decide_pushback, which itself has no
production caller). Signer is bound to the authenticated
caller -- a body signer differing from the caller is rejected
-- and executor/coordinator_approver/qa_tester must be
declared and distinct from the caller; missing field,
self-audit, or forged signer raises ValueError, mapped by
_loop_run to HTTP 400 with no side effects. No new route.
Honest limits are in the helper docstring: declared-identity
enforcement only (task rows store no authoritative executor /
coordinator / QA identity; task executor is a backend, not a
person), and audit-fix reaudit verdicts are out of scope
(the fix poster is not necessarily the re-auditor, so binding
signer to the fix poster would be a mis-binding).

(cherry picked from commit 56dce5165f4d08f38fe1730b9eea3b78f90cb97d)

Reconcile-port: handler_lifecycle.py docstring hunk redirected into cloud/server.py audit-verdict handler (target keeps 6-mixin split; enforcement itself is in audit_loop_wiring.py and applied cleanly).

cloud/audit_loop_wiring.py

@@ -127,6 +127,7 @@ import time
 
 from cloud import audit_loop_driver as _driver
 from cloud import audit_wiring as _wiring
+from cloud.audit_pushback import validate_signer_separation as _validate_signer_separation
 from cloud.merge_gate_hard import evaluate_hard_merge_gate as _hard_gate
 
 # ------------------------------------------------------------------
@@ -715,6 +716,45 @@ def _finish_process(conn, ledger_store, task, action,
     }
 
 
+def _require_signer_separation(verdict, user_id):
+    """Verdict 提交时的 signer 分离强制（A3，fail-closed）。
+
+    signer 不取自请求体：它就是认证调用方 ``user_id`` 本人——
+    体里自带 signer 若与调用方不一致即拒（否则调用方可随手
+    写一个“别人”冒充独立审计，字段不等式检查就成了摆设）。
+    executor / coordinator_approver / qa_tester 必须由提交方
+    在 verdict 里显式声明，缺字段或任一与调用方相等即
+    ValueError（HTTP 层 _loop_run 映射为 400），且校验发生在
+    任何循环状态落盘与链写之前，拒掉的提交不留任何痕迹。
+
+    诚实边界：本校验只保证“提交方 ≠ 提交方自己声明的另三
+    方”，不能识破提交方谎报另三方身份——任务行本身不存
+    执行人/协调批准人/QA 测试人的权威身份（任务 executor 列
+    是 local/docker/cloud 执行后端，不是自然人），服务端无
+    从独立来源核对的依据；谎报身份的防范须另立身份溯源，
+    不在本项用字段检查冒充。audit-fix 路径的 reaudit_verdict
+    不走本校验：那里的 HTTP 提交人是修复提交人，未必是重审
+    审计人，把 signer 绑到修复提交人身上是错绑，见交回报告
+    的已知残留；该路径仍受自动化检查 + 完整重审结论约束。
+    """
+    if not isinstance(verdict, dict):
+        raise TypeError("verdict must be a dict")
+    claimed = verdict.get("signer")
+    if claimed is not None and claimed != user_id:
+        raise ValueError(
+            "verdict signer must equal the authenticated caller")
+    audit = {
+        "signer": user_id,
+        "executor": verdict.get("executor"),
+        "coordinator_approver": verdict.get("coordinator_approver"),
+        "qa_tester": verdict.get("qa_tester"),
+    }
+    valid, reasons = _validate_signer_separation(audit)
+    if not valid:
+        raise ValueError(
+            "audit signer separation failed: %s" % ",".join(reasons))
+
+
 def process_audit_verdict(conn, ledger_store, task, verdict, *,
                           quota_check=None, cost_threshold=None,
                           old_impl_ref=None, user_id="system",
@@ -724,9 +764,17 @@ def process_audit_verdict(conn, ledger_store, task, verdict, *,
     The original task's status is never touched here (the
     server handler likewise does not change it): a fail blocks
     merge/release via the chain and derives a fix subtask, it
-    does not roll the completed task back into a re-run."""
+    does not roll the completed task back into a re-run.
+
+    Signer separation (A3) is enforced before anything is
+    written: the caller is the signer, and the verdict must
+    declare executor / coordinator_approver / qa_tester, all
+    distinct from the caller; otherwise ValueError (HTTP 400)
+    and no loop state or ledger row is created.
+    """
     if not isinstance(verdict, dict):
         raise TypeError("verdict must be a dict")
+    _require_signer_separation(verdict, user_id)
     module_id = resolve_module_id(task)
     contract_version = resolve_contract_version(task, verdict)
     state = load_loop_state(conn, module_id, contract_version)

cloud/server.py

@@ -7240,7 +7240,10 @@ class CloudControlPlaneHandler(OAuthMixin, IdentityMemoryMixin, ProjectMixin, Au
         maintainer 以上，与合并权同级；普通 contributor 不能
         凭空给循环喂结论）。体：{module_id, contract_version,
         verdict: {verdict, source, failures, ...},
-        old_impl_ref?}。原任务状态不动；fail 自动派生修复子
+        old_impl_ref?}。signer 分离（A3）：提交方本人即 signer，
+        verdict 须另声明 executor / coordinator_approver /
+        qa_tester 且均与提交方不同，缺字段、自审或体里 signer
+        与调用方不符一律 400 且不落链不推进。原任务状态不动；fail 自动派生修复子
         任务并经链阻塞合并/发布。"""
         body, jerr = self._read_json()
         if jerr:

cloud/tests/test_audit_loop_wiring.py

@@ -58,6 +58,11 @@ def fail_verdict(text="check x failed", source="audit", **extra):
         "failures": [{"check": "ci_green", "detail": text}],
         "passed_items": ["a", "b", "c"],
         "contract_version": VERSION,
+        # A3: signer is the caller (auditor-1 in _verdict); the
+        # other three roles must be declared and distinct.
+        "executor": "executor-1",
+        "coordinator_approver": "coordinator-1",
+        "qa_tester": "qa-1",
     }
     verdict.update(extra)
     return verdict
@@ -71,6 +76,10 @@ def pass_verdict(**extra):
         "failures": [],
         "passed_items": ["a", "b", "c"],
         "contract_version": VERSION,
+        # A3: see fail_verdict.
+        "executor": "executor-1",
+        "coordinator_approver": "coordinator-1",
+        "qa_tester": "qa-1",
     }
     verdict.update(extra)
     return verdict
@@ -141,7 +150,7 @@ class WiringTestCase(unittest.TestCase):
         self.assertTrue(verify_chain(self.store.conn)["ok"])
 
     def test_full_chain_fail_fix_reaudit_pass_unblocks(self):
-        first = self._verdict(fail_verdict())
+        self._verdict(fail_verdict())
         fix_task = {"id": "task-fix-1", "project_id": "proj-1",
                     "module_id": MODULE, "contract_version": VERSION}
         result = wiring.process_fix_completion(
@@ -435,5 +444,76 @@ class WiringTestCase(unittest.TestCase):
             decision["code"], "release_blocked_state_unavailable")
 
 
+class SignerSeparationTestCase(unittest.TestCase):
+    """A3: validate_signer_separation 在提交路径上的真实调用方。
+
+    覆盖：自审（调用方 = 声明的 executor/coordinator/QA 任一）
+    被拒且不落链不落状态；缺字段被拒；体里 signer 与调用方
+    不符被拒；声明三方均与调用方不同时正常提交。此前该校验
+    只活在重审清单函数里（collect_failed_checks，经无生产
+    调用方的 decide_pushback），提交路径上从未执行。
+
+    注意：本类故意不继承 WiringTestCase——继承会使父类全部
+    测试方法在本类下再跑一遍（全量计数虚增），故只复用其
+    setUp 形状，不复用其用例。
+    """
+
+    def setUp(self):
+        self.tmp = tempfile.TemporaryDirectory()
+        self.ledger_path = os.path.join(self.tmp.name, "ledger.db")
+        self.store = LedgerStore(self.ledger_path)
+        self.conn = sqlite3.connect(":memory:")
+        self.task = {"id": "task-parent", "project_id": "proj-1",
+                     "module_id": MODULE, "contract_version": VERSION}
+
+    def tearDown(self):
+        try:
+            self.store.close()
+        except Exception:
+            pass
+        self.conn.close()
+        self.tmp.cleanup()
+
+    def _verdict(self, verdict, **kwargs):
+        kwargs.setdefault("user_id", "auditor-1")
+        return wiring.process_audit_verdict(
+            self.conn, self.store, self.task, verdict, **kwargs)
+
+    def _chain_rows(self):
+        return self.store.conn.execute(
+            "SELECT COUNT(*) FROM audit_ledger").fetchone()[0]
+
+    def test_self_audit_rejected_without_side_effects(self):
+        for field in ("executor", "coordinator_approver", "qa_tester"):
+            with self.subTest(field=field):
+                verdict = fail_verdict(**{field: "auditor-1"})
+                with self.assertRaises(ValueError) as caught:
+                    self._verdict(verdict)
+                self.assertIn("signer", str(caught.exception))
+                self.assertIsNone(
+                    wiring.load_loop_state(self.conn, MODULE, VERSION))
+                self.assertEqual(self._chain_rows(), 0)
+
+    def test_missing_identity_field_rejected(self):
+        verdict = fail_verdict()
+        del verdict["qa_tester"]
+        with self.assertRaises(ValueError):
+            self._verdict(verdict)
+        self.assertIsNone(
+            wiring.load_loop_state(self.conn, MODULE, VERSION))
+        self.assertEqual(self._chain_rows(), 0)
+
+    def test_forged_signer_rejected(self):
+        verdict = fail_verdict(signer="someone-else")
+        with self.assertRaises(ValueError):
+            self._verdict(verdict)
+        self.assertEqual(self._chain_rows(), 0)
+
+    def test_distinct_declared_roles_submit(self):
+        result = self._verdict(pass_verdict())
+        self.assertEqual(result["action_kind"], "pass")
+        self.assertFalse(result["merge_blocked"])
+
+
 if __name__ == "__main__":
     unittest.main()

cloud/tests/test_server_audit_loop.py

@@ -64,6 +64,11 @@ def fail_body(source="audit"):
             "failures": [{"check": "ci_green", "detail": "x"}],
             "passed_items": ["a", "b", "c"],
             "contract_version": VERSION,
+            # A3: signer is the HTTP caller; the other three
+            # roles must be declared and distinct from it.
+            "executor": "executor-1",
+            "coordinator_approver": "coordinator-1",
+            "qa_tester": "qa-1",
         },
     }
 
@@ -79,6 +84,10 @@ def pass_body():
             "failures": [],
             "passed_items": ["a", "b", "c"],
             "contract_version": VERSION,
+            # A3: see fail_body.
+            "executor": "executor-1",
+            "coordinator_approver": "coordinator-1",
+            "qa_tester": "qa-1",
         },
     }
 
@@ -452,6 +461,23 @@ class AuditLoopServerTestCase(unittest.TestCase):
             fail_body(), self.contrib_key)
         self.assertEqual(status, 403)
 
+    def test_verdict_self_audit_rejected_http(self):
+        # A3 HTTP 层：调用方（owner）自审（声明的 executor 即
+        # 调用方本人）回 400，且循环状态未落盘（audit-loop 404）。
+        task = self._completed_task()
+        body = fail_body()
+        body["verdict"]["executor"] = self.owner_id
+        status, _ = self._req(
+            "POST", "/tasks/%s/audit-verdict" % task["id"],
+            body, self.owner_key)
+        self.assertEqual(status, 400)
+        status, _ = self._req(
+            "GET", "/tasks/%s/audit-loop?module_id=%s"
+                   "&contract_version=%s"
+                   % (task["id"], MODULE, VERSION),
+            key=self.owner_key)
+        self.assertEqual(status, 404)
+
 
 if __name__ == "__main__":
     unittest.main()