import base64 import json from pathlib import Path from cryptography.hazmat.primitives import hashes from cryptography.hazmat.primitives.asymmetric import padding from cryptography.hazmat.primitives.serialization import load_pem_private_key from app.core.config import configured_path MANIFEST_PRIVATE_KEY_PATH = str(configured_path("MANIFEST_PRIVATE_KEY_PATH", "keys/manifest_private_key.pem")) def load_manifest_private_key(): key_path = Path(MANIFEST_PRIVATE_KEY_PATH) if not key_path.exists(): raise FileNotFoundError(f"Manifest private key not found: {key_path}") key_data = key_path.read_bytes() return load_pem_private_key(key_data, password=None) def canonical_manifest_bytes(manifest_obj: dict) -> bytes: # 签名之前必须先规范化 JSON:固定字段顺序、去掉空格、排除 signature 字段。 # 否则同一份 Manifest 在不同环境下序列化结果不同,会导致客户端验签失败。 copy = {k: manifest_obj[k] for k in manifest_obj if k != "signature"} return json.dumps(copy, sort_keys=True, separators=(",", ":"), ensure_ascii=False).encode("utf-8") def sign_manifest(manifest_obj: dict) -> str: # 服务端只保存私钥,客户端只分发公钥。 # 客户端能验证 Manifest 确实由服务端签发,但无法伪造新的签名。 json_text = canonical_manifest_bytes(manifest_obj) private_key = load_manifest_private_key() signature = private_key.sign(json_text, padding.PKCS1v15(), hashes.SHA256()) return base64.b64encode(signature).decode("ascii") def canonical_signed_bytes(obj: dict) -> bytes: copy = {k: obj[k] for k in obj if k != "signature"} return json.dumps(copy, sort_keys=True, separators=(",", ":"), ensure_ascii=False).encode("utf-8") def sign_policy(policy_obj: dict) -> tuple[str, str]: payload = canonical_signed_bytes(policy_obj) signature = load_manifest_private_key().sign(payload, padding.PKCS1v15(), hashes.SHA256()) return payload.decode("utf-8"), base64.b64encode(signature).decode("ascii") def sign_device_identity(identity: dict) -> tuple[str, str]: payload = canonical_signed_bytes(identity) signature = load_manifest_private_key().sign(payload, padding.PKCS1v15(), hashes.SHA256()) return payload.decode("utf-8"), base64.b64encode(signature).decode("ascii")