c62156af53
B1: 项目脚手架 + 数据模型 + 租户管理 - Task 1.1: FastAPI 项目脚手架、SQLite + async SQLAlchemy - Task 1.2: 7 个数据模型 (Tenant, TenantConfig, DigitalEmployee, Conversation, Message, KnowledgeBase, Document) - Task 1.3: 租户 CRUD API + LLM 配置（含 API Key AES 加密） B2: 数字员工配置 + LLM Provider 抽象层 - Task 2.1: 数字员工 CRUD API（关联知识库） - Task 2.2: BaseLLMProvider 抽象接口 + OpenAI/Qwen Provider - Task 2.3: Provider 动态实例化 + test-provider 端点 验证: 26 个测试全部通过 Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
34 lines
959 B
Python
34 lines
959 B
Python
import base64
|
|
import os
|
|
from cryptography.fernet import Fernet
|
|
from cryptography.hazmat.primitives import hashes
|
|
from cryptography.hazmat.primitives.kdf.pbkdf2 import PBKDF2HMAC
|
|
|
|
from app.config import settings
|
|
|
|
|
|
def _get_encryption_key() -> bytes:
|
|
kdf = PBKDF2HMAC(
|
|
algorithm=hashes.SHA256(),
|
|
length=32,
|
|
salt=settings.encryption_salt.encode(),
|
|
iterations=480000,
|
|
)
|
|
return base64.urlsafe_b64encode(kdf.derive(settings.secret_key.encode()))
|
|
|
|
|
|
def encrypt_api_key(plaintext: str) -> str:
|
|
if not plaintext:
|
|
return ""
|
|
fernet = Fernet(_get_encryption_key())
|
|
encrypted = fernet.encrypt(plaintext.encode())
|
|
return base64.urlsafe_b64encode(encrypted).decode()
|
|
|
|
|
|
def decrypt_api_key(ciphertext: str) -> str:
|
|
if not ciphertext:
|
|
return ""
|
|
fernet = Fernet(_get_encryption_key())
|
|
encrypted = base64.urlsafe_b64decode(ciphertext.encode())
|
|
return fernet.decrypt(encrypted).decode()
|