lou
58 lines
1.8 KiB
Python
58 lines
1.8 KiB
Python
"""
|
|
密钥管理模块 (keyring)
|
|
模块: 服务端 / 密钥管理
|
|
输入: key_id
|
|
输出: 密钥材料 -> decrypt
|
|
|
|
预共享密钥方案: 客户端与服务端使用同一 keyring (部署时复制),
|
|
key_id 标识密钥, 支持轮换 (新 key_id 用于新上传, 旧密钥保留仅解密)。
|
|
"""
|
|
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
|
|
from settings import KEYRING_PATH
|
|
|
|
|
|
class KeyringError(RuntimeError):
|
|
"""密钥缺失或文件损坏"""
|
|
|
|
|
|
class Keyring:
|
|
def __init__(self, path: str | Path = KEYRING_PATH) -> None:
|
|
self.path = Path(path)
|
|
self._keys = self._load()
|
|
|
|
def _load(self) -> dict[str, bytes]:
|
|
if not self.path.exists():
|
|
return {}
|
|
try:
|
|
data = json.loads(self.path.read_text(encoding="utf-8"))
|
|
except json.JSONDecodeError as e:
|
|
raise KeyringError(f"密钥文件损坏: {self.path}") from e
|
|
return {kid: bytes.fromhex(hex_str) for kid, hex_str in data.items()}
|
|
|
|
def _save(self) -> None:
|
|
self.path.parent.mkdir(parents=True, exist_ok=True)
|
|
self.path.write_text(
|
|
json.dumps({kid: key.hex() for kid, key in self._keys.items()}, indent=2),
|
|
encoding="utf-8",
|
|
)
|
|
os.chmod(self.path, 0o600)
|
|
|
|
def get_key(self, key_id: str) -> bytes:
|
|
"""取密钥, 不存在抛 KeyringError (任务 failed 的根因之一)"""
|
|
try:
|
|
return self._keys[key_id]
|
|
except KeyError:
|
|
raise KeyringError(f"密钥ID '{key_id}' 不存在于服务端 keyring: {self.path}") from None
|
|
|
|
def has_key(self, key_id: str) -> bool:
|
|
return key_id in self._keys
|
|
|
|
def import_key(self, key_id: str, key: bytes) -> None:
|
|
"""导入密钥 (部署时同步客户端 keyring)"""
|
|
self._keys[key_id] = key
|
|
self._save()
|