lou
50 lines
1.6 KiB
Python
50 lines
1.6 KiB
Python
"""
|
|
解密引擎模块 (decrypt)
|
|
模块: 服务端 / 解密引擎
|
|
输入: 密文主体 + 参数
|
|
输出: 明文流 -> storage
|
|
|
|
流式 AES-GCM (cobblestone) 解密, 块级认证即时检出篡改/密钥错误。
|
|
"""
|
|
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
from cryptography.cobblestone import Cobblestone256Decryptor
|
|
from cryptography.exceptions import InvalidTag
|
|
|
|
from settings import CONTEXT
|
|
from keyring import Keyring, KeyringError
|
|
|
|
|
|
class DecryptError(RuntimeError):
|
|
"""解密失败 (密钥错误/数据损坏)"""
|
|
|
|
|
|
class Decryptor:
|
|
def __init__(self, keyring: Keyring) -> None:
|
|
self.keyring = keyring
|
|
|
|
def decrypt(
|
|
self, merged_path: Path, enc_params: dict[str, Any], out_path: Path
|
|
) -> Path:
|
|
"""流式解密 merged -> out_path, 返回明文路径"""
|
|
key = self.keyring.get_key(enc_params["key_id"])
|
|
context = enc_params.get("context", "").encode() or CONTEXT
|
|
try:
|
|
dec = Cobblestone256Decryptor(key, context)
|
|
with open(merged_path, "rb") as src, open(out_path, "wb") as dst:
|
|
while True:
|
|
chunk = src.read(1 << 16)
|
|
if not chunk:
|
|
break
|
|
dst.write(dec.update(chunk)) # 块级认证: 篡改在此抛 InvalidTag
|
|
dst.write(dec.finalize())
|
|
except InvalidTag as e:
|
|
out_path.unlink(missing_ok=True)
|
|
raise DecryptError("解密验证失败: 密钥不匹配或密文被篡改") from e
|
|
except KeyringError as e:
|
|
out_path.unlink(missing_ok=True)
|
|
raise DecryptError(str(e)) from e
|
|
return out_path
|