95 lines
3.5 KiB
Python
95 lines
3.5 KiB
Python
"""
|
||
Модель данных EncryptedContent для хранения результата шифрования контента.
|
||
|
||
Полезно для сериализации, логирования и передачи между подсистемами uploader-bot.
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import base64
|
||
import json
|
||
import logging
|
||
from dataclasses import dataclass, field, asdict
|
||
from datetime import datetime
|
||
from typing import Any, Dict, Optional
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
|
||
@dataclass
|
||
class EncryptedContent:
|
||
"""
|
||
Универсальная переносимая модель зашифрованного контента.
|
||
Все бинарные поля хранятся в Base64 (строки), чтобы быть JSON-совместимыми.
|
||
"""
|
||
content_id: str
|
||
ciphertext_b64: str
|
||
nonce_b64: str
|
||
tag_b64: str
|
||
|
||
# Подпись и открытый ключ подписанта (Ed25519). Могут отсутствовать.
|
||
signature: Optional[str] = None
|
||
signer_pubkey: Optional[str] = None
|
||
|
||
# Пользовательские/системные метаданные (должны совпадать при верификации)
|
||
metadata: Dict[str, Any] = field(default_factory=dict)
|
||
|
||
# Служебная метка времени создания структуры
|
||
created_at: str = field(default_factory=lambda: datetime.utcnow().isoformat())
|
||
|
||
def to_dict(self) -> Dict[str, Any]:
|
||
"""
|
||
Сериализация в словарь (JSON-совместимый).
|
||
"""
|
||
data = asdict(self)
|
||
# Ничего дополнительно не преобразуем — все поля уже JSON-friendly
|
||
return data
|
||
|
||
def to_json(self) -> str:
|
||
"""
|
||
Сериализация в JSON-строку.
|
||
"""
|
||
payload = self.to_dict()
|
||
try:
|
||
return json.dumps(payload, ensure_ascii=False, sort_keys=True)
|
||
except Exception as e:
|
||
logger.error(f"EncryptedContent.to_json serialization error: {e}")
|
||
raise
|
||
|
||
@classmethod
|
||
def from_dict(cls, data: Dict[str, Any]) -> "EncryptedContent":
|
||
"""
|
||
Десериализация из словаря.
|
||
"""
|
||
required = ["content_id", "ciphertext_b64", "nonce_b64", "tag_b64"]
|
||
for f in required:
|
||
if f not in data:
|
||
raise ValueError(f"Missing required field in EncryptedContent: {f}")
|
||
|
||
return cls(
|
||
content_id=data["content_id"],
|
||
ciphertext_b64=data["ciphertext_b64"],
|
||
nonce_b64=data["nonce_b64"],
|
||
tag_b64=data["tag_b64"],
|
||
signature=data.get("signature"),
|
||
signer_pubkey=data.get("signer_pubkey"),
|
||
metadata=data.get("metadata", {}) or {},
|
||
created_at=data.get("created_at") or datetime.utcnow().isoformat(),
|
||
)
|
||
|
||
@classmethod
|
||
def from_crypto_result(cls, crypto_result: Dict[str, Any]) -> "EncryptedContent":
|
||
"""
|
||
Удобный конструктор из результата ContentCipher.encrypt_content()
|
||
"""
|
||
return cls.from_dict(crypto_result)
|
||
|
||
# Вспомогательные методы для работы с бинарными данными (если необходимо)
|
||
def ciphertext_bytes(self) -> bytes:
|
||
return base64.b64decode(self.ciphertext_b64)
|
||
|
||
def nonce_bytes(self) -> bytes:
|
||
return base64.b64decode(self.nonce_b64)
|
||
|
||
def tag_bytes(self) -> bytes:
|
||
return base64.b64decode(self.tag_b64) |