From 145aa2eb5f94917a51678652f206fdc1ae0ab73f Mon Sep 17 00:00:00 2001 From: host Date: Wed, 4 Nov 2026 09:08:47 +0000 Subject: [PATCH] gapfill G5: wire optional ?current_main_head drift check into GET /audit-scorecard (caller-supplied only, no git/subprocess; absent param means no staleness verdict, verdict never rewrites overall); endpoint tests for passed/absent/empty --- cloud/server.py | 36 ++++++- .../test_gapfill_g5_baseline_snapshot.py | 102 ++++++++++++++++++ 2 files changed, 136 insertions(+), 2 deletions(-) create mode 100644 cloud/tests/test_gapfill_g5_baseline_snapshot.py diff --git a/cloud/server.py b/cloud/server.py index 3c2a1b1..8ba70fd 100644 --- a/cloud/server.py +++ b/cloud/server.py @@ -2730,7 +2730,7 @@ class CloudControlPlaneHandler(OAuthMixin, IdentityMemoryMixin, ProjectMixin, Au elif path == "/metrics-report" and method == "GET": self._metrics_report(user_id) elif path == "/audit-scorecard" and method == "GET": - self._audit_scorecard(user_id) + self._audit_scorecard(user_id, query) elif path == "/models" and method == "GET": self._models_overview(user_id) elif path == "/relay/models" and method == "GET": @@ -8697,9 +8697,21 @@ pid 须为整数、address 须为字符串，否则 400（注册表里的 conn.close() metrics = _compute_metrics(records, costs) self._send_json(200, _build_metrics_report(metrics)) - def _audit_scorecard(self, user_id): + def _audit_scorecard(self, user_id, query=None): """只读 7 维审计评分卡报告端点。 + 可选漂移判定（gapfill G5，机制 item 5 端点快照接线）： + 调用方可在查询串里传 ``?current_main_head=<头>``，端点 + 即用 cloud/audit_scorecard.py 的 check_baseline_stale() + 对返回的卡做一次漂移判定，结果放在响应顶层的 + ``baseline_staleness`` 里（stale / usable_as_pass_basis + 等字段原样返回）。该判定绝不改写卡本身的 overall / + overall_pass（旧值保留只为追溯），也绝不挡住查询： + 未传参数时默认不做判定，响应与以前完全一致；传了 + 参数也永远返回 200 + 报告，只是多一个判定字段。 + 生产代码绝不读 git、不起 subprocess 查版本：main 头 + 只能由调用方传入，端点本身不探测版本控制状态。 + 权限：与 /metrics 同款，任意项目成员即可读（不要求 owner）， 但本端点不统计任何项目数据、不写库、不写账本、不改链头、 不触发推回或重审——它只返回评分卡模块的一张历史示例卡。 @@ -8751,6 +8763,26 @@ pid 须为整数、address 须为字符串，否则 400（注册表里的 "untested_items": [], "product_gate_passed": False}, } + # gapfill G5：可选漂移判定。未传参数时不做判定、不标 + # stale，响应与以前一致；传了参数才附加判定字段，且 + # 判定结果绝不改写报告的 overall / overall_pass。 + heads = (query or {}).get("current_main_head") or [] + current = heads[0] if heads else None + if current: + try: + from cloud.audit_scorecard import ( + check_baseline_stale) + report["baseline_staleness"] = check_baseline_stale( + report, current) + except ImportError: # pragma: no cover - 模块不可用时诚实说明 + report["baseline_staleness"] = { + "checked": False, + "reason": "audit_scorecard_module_unavailable", + "stale": True, + "usable_as_pass_basis": False, + "overall_usable_as_pass_basis": False, + "overall_must_not_be_used_as_pass_basis": True, + } self._send_json(200, report) diff --git a/cloud/tests/test_gapfill_g5_baseline_snapshot.py b/cloud/tests/test_gapfill_g5_baseline_snapshot.py new file mode 100644 index 0000000..5256bf1 --- /dev/null +++ b/cloud/tests/test_gapfill_g5_baseline_snapshot.py @@ -0,0 +1,102 @@ +"""gapfill G5：机制 item 5 端点快照接线定向测试。 + +接线方式：GET /audit-scorecard 可选查询串 +``?current_main_head=<头>``，端点用 check_baseline_stale() +对返回的卡做一次漂移判定，结果放在响应顶层 +``baseline_staleness`` 里。锁定点： +- 未传参数时默认不做判定：响应里没有 baseline_staleness， + 其余字段与以前完全一致，查询绝不被挡住（永远 200）。 +- 传了参数时附加判定字段，但判定绝不改写卡本身的 + overall / overall_pass（示例卡 baseline_head 为 None， + 故任何传入的头都判 stale=True、原因 baseline_head_missing， + 这是失败关闭，不是把旧分改写）。 +- 空串参数按未传入处理：不做判定、不标 stale。 +- 端点本身不读 git、不起 subprocess 查版本：main 头只能 + 由调用方传入（本文件只测 HTTP 入参一路，不测任何 + 版本控制探测——check_baseline_stale 本身已在机制基线 + 测试里钉住“只认调用方传入参数”）。 + +只加测试；生产改动在 cloud/server.py 的 _audit_scorecard。 +""" + +import http.client +import json +import os +import sys +import tempfile +import threading +import unittest +from pathlib import Path + +try: + from cloud.auth import AuthStore + from cloud.server import create_server +except ImportError: + sys.path.insert(0, str(Path(__file__).resolve().parents[2])) + from cloud.auth import AuthStore + from cloud.server import create_server + + +class BaselineSnapshotEndpointTestCase(unittest.TestCase): + def setUp(self): + self.tmp = tempfile.TemporaryDirectory() + self.db_path = os.path.join(self.tmp.name, "control.db") + self.auth = AuthStore(self.db_path) + self.owner_id, self.owner_key = self.auth.create_user("owner") + self.server = create_server(self.db_path, "127.0.0.1", 0) + self.port = self.server.server_address[1] + self.thread = threading.Thread( + target=self.server.serve_forever, daemon=True) + self.thread.start() + + def tearDown(self): + self.server.shutdown() + self.server.server_close() + self.thread.join(timeout=5) + try: + self.auth.conn.close() + except Exception: + pass + self.tmp.cleanup() + + def _req(self, path): + conn = http.client.HTTPConnection( + "127.0.0.1", self.port, timeout=10) + try: + conn.request("GET", path, + headers={"X-API-Key": self.owner_key}) + resp = conn.getresponse() + data = resp.read() + finally: + conn.close() + return resp.status, json.loads(data.decode("utf-8")) + + def test_no_param_means_no_staleness_field(self): + status, report = self._req("/audit-scorecard") + self.assertEqual(status, 200) + self.assertEqual(report["overall"], "not_pass") + self.assertNotIn("baseline_staleness", report) + + def test_param_attaches_staleness_without_rewriting_overall(self): + status, report = self._req( + "/audit-scorecard?current_main_head=82d0e84") + self.assertEqual(status, 200) + # 示例卡无冻结基线：显式要求判定时失败关闭为 stale， + # 但卡本身的 overall / overall_pass 原样不动。 + self.assertEqual(report["overall"], "not_pass") + self.assertFalse(report["overall_pass"]) + stale = report["baseline_staleness"] + self.assertTrue(stale["stale"]) + self.assertFalse(stale["usable_as_pass_basis"]) + self.assertTrue(stale["overall_must_not_be_used_as_pass_basis"]) + self.assertIn("baseline_head_missing", stale["reason"]) + self.assertEqual(stale["current_main_head"], "82d0e84") + + def test_empty_param_means_no_staleness_field(self): + status, report = self._req("/audit-scorecard?current_main_head=") + self.assertEqual(status, 200) + self.assertNotIn("baseline_staleness", report) + + +if __name__ == "__main__": + unittest.main()