6a2632da70
将 Clover 从上层产品包旧仓库中独立出来,建立专属版本控制。 当前状态=纵切片端到端已打通(登录→选品→出文出图→审核→下载包), M1文案质量去套路化已验收。此提交作为后续按核销清单逐条修复的基线。 Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
74 lines
2.3 KiB
Python
74 lines
2.3 KiB
Python
"""
|
||
app/core/security.py — JWT + Fernet 工具函数
|
||
FERNET_KEY 走环境变量，绝不进代码库（基石B）。
|
||
"""
|
||
|
||
import logging
|
||
from datetime import datetime, timedelta, timezone
|
||
from typing import Any
|
||
|
||
import jwt
|
||
from cryptography.fernet import Fernet, InvalidToken
|
||
|
||
from app.core.config import get_settings
|
||
|
||
logger = logging.getLogger(__name__)
|
||
settings = get_settings()
|
||
|
||
# ── Fernet 加密（API Key 存取）──────────────────────────
|
||
_fernet: Fernet | None = None
|
||
|
||
|
||
def _get_fernet() -> Fernet:
|
||
global _fernet
|
||
if _fernet is None:
|
||
_fernet = Fernet(settings.FERNET_KEY.encode())
|
||
return _fernet
|
||
|
||
|
||
def encrypt_api_key(plain_key: str) -> str:
|
||
"""加密 API Key，返回密文字符串。绝不打印 plain_key。"""
|
||
return _get_fernet().encrypt(plain_key.encode()).decode()
|
||
|
||
|
||
def decrypt_api_key(encrypted_key: str) -> str:
|
||
"""
|
||
解密 API Key。只在 Celery worker 内部调用。
|
||
解密结果只活在调用函数的局部变量，不落盘、不打日志。
|
||
"""
|
||
try:
|
||
return _get_fernet().decrypt(encrypted_key.encode()).decode()
|
||
except InvalidToken:
|
||
logger.error("Fernet decrypt failed: invalid token (key may be rotated)")
|
||
raise ValueError("API key decryption failed")
|
||
|
||
|
||
# ── JWT ──────────────────────────────────────────────────
|
||
def create_access_token(
|
||
user_id: int,
|
||
workspace_id: int,
|
||
role: str,
|
||
expires_delta: timedelta | None = None,
|
||
) -> str:
|
||
expire = datetime.now(timezone.utc) + (
|
||
expires_delta or timedelta(minutes=settings.JWT_EXPIRE_MINUTES)
|
||
)
|
||
payload: dict[str, Any] = {
|
||
"sub": str(user_id),
|
||
"user_id": user_id,
|
||
"current_workspace_id": workspace_id,
|
||
"role": role,
|
||
"exp": expire,
|
||
}
|
||
return jwt.encode(payload, settings.JWT_SECRET, algorithm="HS256")
|
||
|
||
|
||
def decode_access_token(token: str) -> dict[str, Any]:
|
||
"""验签并返回 payload。无效则抛 jwt.PyJWTError。"""
|
||
return jwt.decode(token, settings.JWT_SECRET, algorithms=["HS256"])
|
||
|
||
|
||
def mask_api_key(plain_key: str) -> str:
|
||
"""只返回后4位（展示用），不暴露完整 key。"""
|
||
return plain_key[-4:] if len(plain_key) >= 4 else "****"
|