diff --git a/core/template-console/push_channel.py b/core/template-console/push_channel.py new file mode 100644 index 0000000..c255e49 --- /dev/null +++ b/core/template-console/push_channel.py @@ -0,0 +1,263 @@ +# -*- coding: utf-8 -*- +"""⑤.7 配置台 ↔ 内核配置推送契约 —— issue #67 / PRD ⑤.7。 + +发布(#66)之后的下一步是**把配置真正送进内核**让 edge-gateway / rag-kb / +model-framework 生效。配置台与内核是两个独立部署单元,二者通过**配置推送契约** +解耦:配置台把一份已发布版本打包为内核可消费的 **JSON manifest**(带校验和), +内核侧拉取/接收后先验完整性再加载。本模块实现这个契约的"配置台侧": + +- ``PushManifest``:推送给内核的清单(版本 / 快照 / 校验和 / 生成时间 / 来源); +- ``PushChannel``:推送通道。 + - ``build_manifest(release)``:把 Release 打包成 manifest,计算 SHA256 校验和 + (对快照做规范 JSON 序列化后哈希,确保内核侧可复算验证); + - ``push(release)``:模拟推送——把 manifest 写到内核预期的接收目录 + (``/manifest-.json``),并记录推送日志(幂等:同版本不重复推送); + - ``pushed_versions()``:已成功推送的版本清单; + - ``verify(manifest)``:校验 manifest 的校验和是否一致(内核侧或配置台侧复用)。 + +幂等性:同一版本重复 push 返回已推送的旧记录(不覆盖、不重复写文件),避免内核 +重复加载;要重推需先 ``retract``(撤回)该版本。 + +零运行时依赖:仅用 json / hashlib / dataclass / 标准库。 +""" +from __future__ import annotations + +import hashlib +import json +import os +from dataclasses import asdict, dataclass, field +from datetime import datetime, timezone +from typing import Any, Dict, List, Optional + +from .release import Release + + +MANIFEST_SCHEMA_VERSION = 1 +MANIFEST_FILENAME_FMT = "manifest-{version}.json" +PUSH_LOG_FILENAME = "push_log.json" +PUSH_LOG_SCHEMA_VERSION = 1 + + +def _now_iso() -> str: + return datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ") + + +def _canonical_json(obj: Any) -> str: + """规范 JSON 序列化(排序键、无空白),用于稳定哈希。""" + return json.dumps(obj, sort_keys=True, ensure_ascii=False, separators=(",", ":")) + + +def checksum(snapshot: Dict[str, Any]) -> str: + """计算配置快照的 SHA256 校验和(规范序列化后哈希)。 + + 内核侧收到 manifest 后,对 ``snapshot`` 用同样算法复算,比对 ``checksum`` + 即可确认传输无损/未篡改。 + """ + return hashlib.sha256(_canonical_json(snapshot).encode("utf-8")).hexdigest() + + +# --------------------------------------------------------------------------- +# 推送清单 +# --------------------------------------------------------------------------- + +@dataclass +class PushManifest: + """推送给内核的配置清单(自描述:版本/快照/校验和/来源)。""" + + schema_version: int = MANIFEST_SCHEMA_VERSION + version: str = "" # 对应 Release 的 semver + snapshot: Dict[str, Any] = field(default_factory=dict) + checksum: str = "" # snapshot 的 SHA256 + generated_at: str = "" # manifest 生成时间 + source: str = "template-console" # 来源标识(内核侧据此识别推送方) + description: str = "" # 推送说明(可解释) + + def to_dict(self) -> dict: + return asdict(self) + + def to_json(self) -> str: + """manifest 序列化为 JSON 文本(推送载荷)。""" + return json.dumps(self.to_dict(), ensure_ascii=False, indent=2) + + @classmethod + def from_dict(cls, raw: dict) -> "PushManifest": + return cls( + schema_version=raw.get("schema_version", MANIFEST_SCHEMA_VERSION), + version=raw["version"], + snapshot=raw.get("snapshot", {}), + checksum=raw.get("checksum", ""), + generated_at=raw.get("generated_at", ""), + source=raw.get("source", "template-console"), + description=raw.get("description", ""), + ) + + +# --------------------------------------------------------------------------- +# 推送通道 +# --------------------------------------------------------------------------- + +@dataclass +class PushRecord: + """一次推送的记录(幂等判定与审计依据)。""" + + version: str + checksum: str + pushed_at: str + pushed_by: str + manifest_path: str + status: str = "pushed" # pushed / retracted + reason: str = "" + + +class PushChannel: + """配置台 → 内核的配置推送通道(基于文件系统的模拟推送)。 + + 用法: + rm = ReleaseManager(store) + rel = rm.publish("1.0.0", ...) + ch = PushChannel(inbox="/path/to/kernel/inbox") + manifest = ch.push(rel, pushed_by="admin") + # 内核侧:读 manifest,复算 checksum 比对,加载 snapshot + """ + + def __init__(self, inbox: str, push_log_path: Optional[str] = None) -> None: + """``inbox`` 是内核侧接收目录(模拟推送就是把 manifest 写到此处)。 + + ``push_log_path`` 推送日志路径(默认与 inbox 同目录的 push_log.json), + 记录每个版本的推送状态,支撑幂等与撤回。 + """ + self.inbox = inbox + os.makedirs(inbox, exist_ok=True) + self.push_log_path = push_log_path or os.path.join(inbox, PUSH_LOG_FILENAME) + + # -- manifest 构建 -- + def build_manifest( + self, release: Release, description: str = "", + ) -> PushManifest: + """把 Release 打包为 PushManifest(含校验和)。""" + snap = release.snapshot + return PushManifest( + schema_version=MANIFEST_SCHEMA_VERSION, + version=release.version, + snapshot=snap, + checksum=checksum(snap), + generated_at=_now_iso(), + source="template-console", + description=description or f"推送版本 {release.version}", + ) + + # -- 推送日志 -- + def _read_log(self) -> List[PushRecord]: + if not os.path.isfile(self.push_log_path): + return [] + with open(self.push_log_path, "r", encoding="utf-8") as fh: + blob = json.load(fh) + return [PushRecord(**r) for r in blob.get("records", [])] + + def _write_log(self, records: List[PushRecord]) -> None: + blob = { + "schema_version": PUSH_LOG_SCHEMA_VERSION, + "records": [asdict(r) for r in records], + } + tmp = self.push_log_path + ".tmp" + with open(tmp, "w", encoding="utf-8") as fh: + json.dump(blob, fh, ensure_ascii=False, indent=2) + os.replace(tmp, self.push_log_path) + + def _find_record(self, version: str) -> Optional[PushRecord]: + for r in self._read_log(): + if r.version == version: + return r + return None + + # -- 推送 / 撤回 -- + def push( + self, release: Release, pushed_by: str = "system", + description: str = "", force: bool = False, + ) -> PushManifest: + """推送一个已发布版本到内核接收目录(幂等:同版本不重复推送)。 + + 幂等性:若该版本已成功推送且未撤回,直接返回原 manifest(不重复写文件、 + 不重复触发内核加载)。要强制重推,先 ``retract`` 或传 ``force=True``。 + + Args: + release: 已发布的版本(含快照); + pushed_by: 推送人(对接 RBAC); + description: 推送说明; + force: 强制重推(覆盖既有 manifest)。 + + Returns: + 推送的 PushManifest。 + """ + existing = self._find_record(release.version) + if existing and existing.status == "pushed" and not force: + # 幂等:返回已推送的 manifest(从 inbox 读回) + if os.path.isfile(existing.manifest_path): + with open(existing.manifest_path, "r", encoding="utf-8") as fh: + return PushManifest.from_dict(json.load(fh)) + + manifest = self.build_manifest(release, description=description) + manifest_path = os.path.join( + self.inbox, MANIFEST_FILENAME_FMT.format(version=release.version)) + tmp = manifest_path + ".tmp" + with open(tmp, "w", encoding="utf-8") as fh: + fh.write(manifest.to_json()) + os.replace(tmp, manifest_path) + + # 更新推送日志(覆盖同版本旧记录) + records = [r for r in self._read_log() if r.version != release.version] + records.append(PushRecord( + version=release.version, checksum=manifest.checksum, + pushed_at=_now_iso(), pushed_by=pushed_by, + manifest_path=manifest_path, status="pushed", + reason=description or f"推送 {release.version}", + )) + self._write_log(records) + return manifest + + def retract(self, version: str, by: str = "system", reason: str = "") -> bool: + """撤回一个已推送版本(标记为 retracted,不删 manifest 文件,可追溯)。 + + 撤回后该版本可重新 push(幂等解除)。返回是否实际撤回。 + """ + rec = self._find_record(version) + if rec is None or rec.status != "pushed": + return False + records = self._read_log() + for i, r in enumerate(records): + if r.version == version: + records[i] = PushRecord( + version=r.version, checksum=r.checksum, + pushed_at=r.pushed_at, pushed_by=r.pushed_by, + manifest_path=r.manifest_path, status="retracted", + reason=f"撤回 by {by}:{reason or '未说明'}", + ) + self._write_log(records) + return True + + # -- 查询 / 校验 -- + def pushed_versions(self) -> List[Dict[str, Any]]: + """已推送版本摘要(配置台推送状态列表用)。""" + return [ + {"version": r.version, "checksum": r.checksum, + "pushed_at": r.pushed_at, "pushed_by": r.pushed_by, + "status": r.status} + for r in self._read_log() + ] + + @staticmethod + def verify(manifest: PushManifest) -> bool: + """校验 manifest 的 checksum 与其 snapshot 是否一致。 + + 内核侧收到 manifest 后调用此方法,确认传输无损;配置台侧也可在推送前自检。 + """ + return manifest.checksum == checksum(manifest.snapshot) + + @staticmethod + def verify_payload(payload: Dict[str, Any]) -> bool: + """从原始 payload(dict)校验:用同算法复算 checksum 比对。""" + try: + manifest = PushManifest.from_dict(payload) + except (KeyError, TypeError): + return False + return PushChannel.verify(manifest) diff --git a/core/template-console/tests/test_push_channel.py b/core/template-console/tests/test_push_channel.py new file mode 100644 index 0000000..aa8a244 --- /dev/null +++ b/core/template-console/tests/test_push_channel.py @@ -0,0 +1,215 @@ +# -*- coding: utf-8 -*- +"""配置推送契约测试(issue #67)。 + +覆盖: +1. manifest 构建(版本/快照/校验和/来源); +2. checksum 稳定性 + 完整性校验(verify); +3. 推送幂等(同版本不重复写文件、返回原 manifest); +4. force 强制重推; +5. retract 撤回 + 重新推送; +6. 模拟传输损坏(篡改 snapshot → verify 失败); +7. 推送日志与查询。 +""" +import json +import os +import sys +import tempfile +import unittest + +sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) +import _bootstrap # noqa: F401 + +from template_console.config_store import ConfigKind, ConfigStore # noqa: E402 +from template_console.push_channel import ( # noqa: E402 + PushChannel, + PushManifest, + PushRecord, + checksum, +) +from template_console.release import Release, ReleaseManager # noqa: E402 + + +def _make_release(version: str = "1.0.0") -> Release: + """构造一个带快照的 Release(不走文件系统,直接内存构造)。""" + snap = { + "schema_version": 1, + "captured_at": "2026-01-01T00:00:00Z", + "kinds": {"model_param": [ + {"key": "lr", "value": 0.001, "kind": "model_param", + "meaning": "学习率", "updated_by": "li", "reason": "init", + "updated_at": "2026-01-01T00:00:00Z"}]}, + } + return Release(version=version, created_at="2026-01-01T00:00:00Z", + snapshot=snap, released_by="admin", changelog="t", reason="r") + + +class _Tmp: + def __init__(self): + self._tmp = tempfile.mkdtemp() + self.inbox = os.path.join(self._tmp, "inbox") + + def cleanup(self): + import shutil + shutil.rmtree(self._tmp, ignore_errors=True) + + +class ChecksumTest(unittest.TestCase): + """校验和稳定性 + 完整性。""" + + def test_checksum_stable(self): + snap = {"kinds": {"a": [1, 2, 3]}} + self.assertEqual(checksum(snap), checksum(snap)) + + def test_checksum_key_order_independent(self): + # 键序不影响校验和(规范序列化) + a = checksum({"x": 1, "y": 2}) + b = checksum({"y": 2, "x": 1}) + self.assertEqual(a, b) + + def test_checksum_changes_on_value_change(self): + self.assertNotEqual(checksum({"v": 1}), checksum({"v": 2})) + + def test_checksum_is_sha256_hex(self): + cs = checksum({"v": 1}) + self.assertEqual(len(cs), 64) + self.assertTrue(all(c in "0123456789abcdef" for c in cs)) + + +class ManifestTest(unittest.TestCase): + """manifest 构建。""" + + def test_build_manifest_has_checksum(self): + ch = PushChannel(inbox=tempfile.mkdtemp()) + rel = _make_release() + m = ch.build_manifest(rel) + self.assertEqual(m.version, "1.0.0") + self.assertTrue(m.checksum) + self.assertEqual(m.source, "template-console") + self.assertTrue(m.generated_at) + + def test_manifest_roundtrip(self): + m = PushManifest(version="1.0.0", snapshot={"a": 1}, + checksum=checksum({"a": 1}), generated_at="t") + text = m.to_json() + m2 = PushManifest.from_dict(json.loads(text)) + self.assertEqual(m2.version, "1.0.0") + self.assertEqual(m2.checksum, m.checksum) + + +class PushIdempotencyTest(unittest.TestCase): + """推送幂等。""" + + def setUp(self): + self.ctx = _Tmp() + self.ch = PushChannel(inbox=self.ctx.inbox) + self.rel = _make_release() + + def tearDown(self): + self.ctx.cleanup() + + def test_push_writes_manifest_file(self): + self.ch.push(self.rel, pushed_by="admin") + path = os.path.join(self.ctx.inbox, "manifest-1.0.0.json") + self.assertTrue(os.path.isfile(path)) + + def test_push_is_idempotent(self): + m1 = self.ch.push(self.rel, pushed_by="a") + m2 = self.ch.push(self.rel, pushed_by="b") # 重复推送 + # 同版本返回同一 manifest(校验和一致) + self.assertEqual(m1.checksum, m2.checksum) + # 推送日志只有一条记录 + self.assertEqual(len(self.ch.pushed_versions()), 1) + + def test_force_overrides_idempotency(self): + self.ch.push(self.rel, pushed_by="a") + before = self.ch.pushed_versions()[0]["pushed_at"] + # force 重推(时间戳可能更新) + self.ch.push(self.rel, pushed_by="b", force=True) + records = self.ch.pushed_versions() + self.assertEqual(len(records), 1) + self.assertEqual(records[0]["pushed_by"], "b") + + def test_push_log_records_pushed_by(self): + self.ch.push(self.rel, pushed_by="admin_zhang") + rec = self.ch.pushed_versions()[0] + self.assertEqual(rec["pushed_by"], "admin_zhang") + self.assertEqual(rec["status"], "pushed") + + +class RetractTest(unittest.TestCase): + """撤回 + 重新推送。""" + + def setUp(self): + self.ctx = _Tmp() + self.ch = PushChannel(inbox=self.ctx.inbox) + self.rel = _make_release() + + def tearDown(self): + self.ctx.cleanup() + + def test_retract_marks_status(self): + self.ch.push(self.rel) + self.assertTrue(self.ch.retract("1.0.0", by="admin", reason="有问题")) + rec = self.ch.pushed_versions()[0] + self.assertEqual(rec["status"], "retracted") + + def test_retract_unknown_returns_false(self): + self.assertFalse(self.ch.retract("9.9.9")) + + def test_retract_allows_repush(self): + self.ch.push(self.rel) + self.ch.retract("1.0.0") + # 撤回后可重新推送(幂等解除) + m = self.ch.push(self.rel, pushed_by="admin2") + rec = self.ch.pushed_versions()[0] + self.assertEqual(rec["status"], "pushed") + self.assertEqual(rec["pushed_by"], "admin2") + + +class VerifyTest(unittest.TestCase): + """完整性校验。""" + + def test_verify_valid_manifest(self): + ch = PushChannel(inbox=tempfile.mkdtemp()) + m = ch.build_manifest(_make_release()) + self.assertTrue(PushChannel.verify(m)) + + def test_verify_tampered_snapshot_fails(self): + ch = PushChannel(inbox=tempfile.mkdtemp()) + m = ch.build_manifest(_make_release()) + # 篡改 snapshot 但不改 checksum → 校验失败 + m.snapshot["kinds"]["model_param"][0]["value"] = 0.999 + self.assertFalse(PushChannel.verify(m)) + + def test_verify_payload_dict(self): + ch = PushChannel(inbox=tempfile.mkdtemp()) + m = ch.build_manifest(_make_release()) + self.assertTrue(PushChannel.verify_payload(m.to_dict())) + + def test_verify_payload_bad_dict(self): + self.assertFalse(PushChannel.verify_payload({"nope": 1})) + + +class IntegrationTest(unittest.TestCase): + """端到端:store → publish → push → verify。""" + + def test_store_publish_push_flow(self): + tmp = tempfile.mkdtemp() + try: + store = ConfigStore(tmp) + store.upsert(ConfigKind.MODEL_PARAM, "lr", 0.001) + rm = ReleaseManager(store) + rel = rm.publish("1.0.0", released_by="admin", changelog="首发") + inbox = os.path.join(tmp, "inbox") + ch = PushChannel(inbox=inbox) + m = ch.push(rel, pushed_by="admin") + # 内核侧校验通过 + self.assertTrue(PushChannel.verify(m)) + self.assertEqual(len(ch.pushed_versions()), 1) + finally: + import shutil + shutil.rmtree(tmp, ignore_errors=True) + + +if __name__ == "__main__": + unittest.main()