264 lines
10 KiB
Python
264 lines
10 KiB
Python
# -*- 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 写到内核预期的接收目录
|
||
（``<inbox>/manifest-<version>.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)
|