83 lines
3.0 KiB
Python
83 lines
3.0 KiB
Python
"""魂籍（portable_identity_record）：可验证、版本化的身份与状态包。
|
|
|
|
完整性模型：
|
|
- payload 规范化序列化（排序键、紧凑分隔符）后取 SHA-256 摘要；
|
|
- 签名 = HMAC-SHA256(密钥, "{character_id}|{version}|{digest}")，绑定人物与版本；
|
|
- 校验：重算摘要与签名比对。篡改 payload 破坏摘要；把记录复制到别的人物/版本，
|
|
签名不匹配——文件相同或人物 ID 相同不足以冒充原主体（data-model.md §1）。
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import hmac
|
|
import json
|
|
|
|
from app.core.config import settings
|
|
from app.models.jingsheng import IdentityRecord
|
|
|
|
|
|
def _canonical(payload: dict) -> str:
|
|
return json.dumps(payload, ensure_ascii=False, sort_keys=True, separators=(",", ":"))
|
|
|
|
|
|
def _digest(payload: dict) -> str:
|
|
return hashlib.sha256(_canonical(payload).encode("utf-8")).hexdigest()
|
|
|
|
|
|
def _signature(character_id: str, version: int, digest: str) -> str:
|
|
secret = settings.identity_record_secret
|
|
message = f"{character_id}|{version}|{digest}".encode("utf-8")
|
|
return hmac.new(secret.encode("utf-8"), message, hashlib.sha256).hexdigest()
|
|
|
|
|
|
def build_identity_record(
|
|
*,
|
|
character_id: str,
|
|
version: int,
|
|
identity: dict,
|
|
runtime_state: dict,
|
|
data_state: dict | None = None,
|
|
) -> IdentityRecord:
|
|
"""构建一条魂籍版本记录（调用方负责落库）。"""
|
|
payload = {
|
|
"spec": "hunjing.portable_identity_record.v1",
|
|
"character_id": character_id,
|
|
"version": version,
|
|
"identity": identity,
|
|
"data_state": data_state or {"note": "数据状态由各域自报（M1 占位）"},
|
|
"runtime_state": runtime_state,
|
|
}
|
|
digest = _digest(payload)
|
|
return IdentityRecord(
|
|
character_id=character_id,
|
|
version=version,
|
|
payload_json=json.dumps(payload, ensure_ascii=False),
|
|
digest=digest,
|
|
signature=_signature(character_id, version, digest),
|
|
)
|
|
|
|
|
|
class IdentityVerificationError(Exception):
|
|
"""校验失败：摘要或签名不匹配。"""
|
|
|
|
def __init__(self, reason: str) -> None:
|
|
super().__init__(reason)
|
|
self.reason = reason
|
|
|
|
|
|
def verify_identity_record(record: IdentityRecord) -> dict:
|
|
"""校验魂籍记录，返回规范化 payload；失败抛 IdentityVerificationError。"""
|
|
try:
|
|
payload = json.loads(record.payload_json)
|
|
except json.JSONDecodeError as exc:
|
|
raise IdentityVerificationError("payload 无法解析") from exc
|
|
if _digest(payload) != record.digest:
|
|
raise IdentityVerificationError("摘要不匹配（payload 已被篡改）")
|
|
expected = _signature(record.character_id, record.version, record.digest)
|
|
if not hmac.compare_digest(expected, record.signature):
|
|
raise IdentityVerificationError("签名不匹配（非本平台签发或被移植冒用）")
|
|
if payload.get("character_id") != record.character_id or payload.get("version") != record.version:
|
|
raise IdentityVerificationError("payload 声明的身份与记录不符")
|
|
return payload
|