GuangYu/QuestionProject
4
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: 加密并隐藏系统敏感配置

Browse Source
This commit is contained in:
Nelson
2026-07-10 11:49:21 +08:00
parent aedecb3aab
commit 9fb3dad5bd
7 changed files with 92 additions and 17 deletions
Download Patch File Download Diff File Expand all files Collapse all files

21
ai_knowledge_base_v2/apps/backend/app/api/admin_settings.py
View File

@@ -22,6 +22,7 @@ from app.services.feishu_service import FeishuKnowledgeService
from app.services.knowledge_service import KnowledgeScope
from app.services.model_service import ModelClientService
from app.services.rag_service import RagResult
from app.services.secret_service import MASKED_SECRET, SENSITIVE_CONFIG_KEYS, SecretService
router = APIRouter()
@@ -110,7 +111,7 @@ def create_model(
model_name=payload.modelName,
base_url=payload.baseUrl,
api_url=payload.apiUrl,
api_key=payload.apiKey,
api_key=SecretService.encrypt(payload.apiKey),
auth_type=payload.authType,
api_version=payload.apiVersion,
temperature=payload.temperature,
@@ -152,8 +153,8 @@ def update_model(
model.model_name = payload.modelName
model.base_url = payload.baseUrl
model.api_url = payload.apiUrl
if payload.apiKey:
model.api_key = payload.apiKey
if payload.apiKey and payload.apiKey != MASKED_SECRET:
model.api_key = SecretService.encrypt(payload.apiKey)
model.auth_type = payload.authType
model.api_version = payload.apiVersion
model.temperature = payload.temperature
@@ -237,7 +238,12 @@ def save_config(
config = db.scalar(select(SystemConfig).where(SystemConfig.config_key == payload.configKey))
if config is None:
config = SystemConfig(config_key=payload.configKey, config_value=payload.configValue)
config.config_value = payload.configValue
if payload.configKey in SENSITIVE_CONFIG_KEYS:
if payload.configValue == MASKED_SECRET:
return api_success(_config_dict(config))
config.config_value = SecretService.encrypt(payload.configValue)
else:
config.config_value = payload.configValue
config.description = payload.description
config.updated_by = current_admin.id
db.add(config)
@@ -257,7 +263,7 @@ def _model_dict(model: ModelConfig) -> dict:
"modelName": model.model_name,
"baseUrl": model.base_url,
"apiUrl": model.api_url,
"apiKeyMasked": "******" if model.api_key else "",
"apiKeyMasked": SecretService.masked(model.api_key),
"authType": model.auth_type,
"apiVersion": model.api_version,
"temperature": float(model.temperature) if model.temperature is not None else None,
@@ -309,10 +315,13 @@ def clear_feishu_cache(current_admin: Admin = Depends(get_current_admin)) -> dic
def _config_dict(config: SystemConfig) -> dict:
value = config.config_value
if config.config_key in SENSITIVE_CONFIG_KEYS:
value = SecretService.masked(value)
return {
"id": config.id,
"configKey": config.config_key,
"configValue": config.config_value,
"configValue": value,
"description": config.description,
"updatedAt": config.updated_at,
}

1
ai_knowledge_base_v2/apps/backend/app/core/config.py
View File

@@ -33,6 +33,7 @@ class Settings(BaseSettings):
auto_create_tables: bool = False
jwt_secret_key: str = "change-me-in-production"
config_encryption_key: str = ""
jwt_algorithm: str = "HS256"
access_token_expire_minutes: int = 60 * 24 * 30

2
ai_knowledge_base_v2/apps/backend/app/main.py
View File

@@ -9,11 +9,13 @@ from app.api.router import api_router
from app.core.config import get_settings
from app.core.database import create_tables
from app.core.exception_handlers import register_exception_handlers
from app.services.secret_service import SecretService
@asynccontextmanager
async def lifespan(app: FastAPI):
settings = get_settings()
SecretService.validate_production_key()
if settings.auto_create_tables:
create_tables()
yield

3
ai_knowledge_base_v2/apps/backend/app/services/feishu_service.py
View File

@@ -13,6 +13,7 @@ from app.core.config import get_settings
from app.models.ai_config import SystemConfig
from app.services.external_errors import ExternalServiceError
from app.services.knowledge_service import KnowledgeScope
from app.services.secret_service import SecretService
if TYPE_CHECKING:
from app.services.rag_service import RetrievedChunk
@@ -812,7 +813,7 @@ def _feishu_retrieval_config(db: Session | None) -> FeishuRetrievalConfig:
return FeishuRetrievalConfig(
search_url=_config_text(db, "feishu_search_url", settings.feishu_search_url),
app_id=_config_text(db, "feishu_app_id", settings.feishu_app_id),
app_secret=_config_text(db, "feishu_app_secret", settings.feishu_app_secret),
app_secret=SecretService.decrypt(_config_text(db, "feishu_app_secret", settings.feishu_app_secret)),
timeout_seconds=_config_int(db, "feishu_timeout_seconds", settings.feishu_timeout_seconds, minimum=1, maximum=120),
retry_count=_config_int(db, "feishu_retry_count", settings.feishu_retry_count, minimum=0, maximum=10),
)

17
ai_knowledge_base_v2/apps/backend/app/services/model_service.py
View File

@@ -13,6 +13,7 @@ from app.core.config import get_settings
from app.models.ai_config import ModelConfig, SystemConfig
from app.services.external_errors import ExternalServiceError
from app.services.rag_service import NO_HIT_ANSWER, RagResult
from app.services.secret_service import SecretService
@dataclass(frozen=True)
@@ -343,7 +344,7 @@ def _call_minimax(model: ModelConfig, rag_result: RagResult) -> str:
_put_if_not_none(payload, "temperature", _decimal_to_float(model.temperature))
_put_if_not_none(payload, "top_p", _decimal_to_float(model.top_p))
headers = {"Authorization": f"Bearer {model.api_key}", "Content-Type": "application/json"}
headers = {"Authorization": f"Bearer {_model_api_key(model)}", "Content-Type": "application/json"}
try:
response = httpx.post(base_url, json=payload, headers=headers, timeout=model.timeout_second)
@@ -398,16 +399,16 @@ def _resolve_gemini_endpoint(model: ModelConfig) -> str:
if model.api_url:
return model.api_url
base_url = (model.base_url or "https://generativelanguage.googleapis.com/v1beta").rstrip("/")
query = urlencode({"key": model.api_key})
query = urlencode({"key": _model_api_key(model)})
return f"{base_url}/models/{model.model_name}:generateContent?{query}"
def _auth_headers(model: ModelConfig) -> dict[str, str]:
headers = {"Content-Type": "application/json"}
if model.auth_type == "api_key":
headers["x-api-key"] = model.api_key
headers["x-api-key"] = _model_api_key(model)
else:
headers["Authorization"] = f"Bearer {model.api_key}"
headers["Authorization"] = f"Bearer {_model_api_key(model)}"
if model.api_version:
headers["api-version"] = model.api_version
return headers
@@ -416,9 +417,9 @@ def _auth_headers(model: ModelConfig) -> dict[str, str]:
def _anthropic_headers(model: ModelConfig) -> dict[str, str]:
headers = {"Content-Type": "application/json"}
if model.auth_type == "bearer":
headers["Authorization"] = f"Bearer {model.api_key}"
headers["Authorization"] = f"Bearer {_model_api_key(model)}"
else:
headers["x-api-key"] = model.api_key
headers["x-api-key"] = _model_api_key(model)
if model.api_version:
headers["anthropic-version"] = model.api_version
elif model.provider == "anthropic":
@@ -430,6 +431,10 @@ def _decimal_to_float(value: Any) -> float | None:
return float(value) if value is not None else None
def _model_api_key(model: ModelConfig) -> str:
return SecretService.decrypt(model.api_key)
def _put_if_not_none(target: dict[str, Any], key: str, value: Any) -> None:
if value is not None:
target[key] = value

58
ai_knowledge_base_v2/apps/backend/app/services/secret_service.py Normal file
View File

@@ -0,0 +1,58 @@
from __future__ import annotations
import base64
import hashlib
from cryptography.fernet import Fernet, InvalidToken
from app.core.config import get_settings
ENCRYPTED_PREFIX = "enc:v1:"
MASKED_SECRET = "******"
SENSITIVE_CONFIG_KEYS = {"aliyun_sms_access_key_secret", "feishu_app_secret"}
class SecretService:
@staticmethod
def encrypt(value: str) -> str:
if not value or value.startswith(ENCRYPTED_PREFIX):
return value
token = SecretService._fernet().encrypt(value.encode("utf-8")).decode("ascii")
return f"{ENCRYPTED_PREFIX}{token}"
@staticmethod
def decrypt(value: str | None) -> str:
if not value:
return ""
if not value.startswith(ENCRYPTED_PREFIX):
return value
try:
return SecretService._fernet().decrypt(value[len(ENCRYPTED_PREFIX) :].encode("ascii")).decode("utf-8")
except InvalidToken as exc:
raise RuntimeError("敏感配置无法解密，请检查 CONFIG_ENCRYPTION_KEY 是否与保存时一致") from exc
@staticmethod
def masked(value: str | None) -> str:
return MASKED_SECRET if value else ""
@staticmethod
def validate_production_key() -> None:
settings = get_settings()
if settings.app_env.strip().lower() in {"local", "dev", "development", "docker", "test", "testing"}:
return
if not settings.config_encryption_key.strip():
raise RuntimeError("生产环境必须配置 CONFIG_ENCRYPTION_KEY")
SecretService._fernet()
@staticmethod
def _fernet() -> Fernet:
settings = get_settings()
configured = settings.config_encryption_key.strip()
if configured:
try:
return Fernet(configured.encode("ascii"))
except (ValueError, TypeError) as exc:
raise RuntimeError("CONFIG_ENCRYPTION_KEY 格式无效，必须是 Fernet 密钥") from exc
fallback = base64.urlsafe_b64encode(hashlib.sha256(settings.jwt_secret_key.encode("utf-8")).digest())
return Fernet(fallback)

7
ai_knowledge_base_v2/apps/backend/app/services/sms_code_service.py
View File

@@ -14,6 +14,7 @@ from sqlalchemy.orm import Session
from app.core.config import get_settings
from app.models.ai_config import SystemConfig
from app.services.redis_client import get_sync_redis_client
from app.services.secret_service import SecretService
logger = logging.getLogger(__name__)
@@ -102,10 +103,8 @@ def _load_sms_config(db: Session) -> SmsProviderConfig:
"aliyun_sms_access_key_id",
settings.aliyun_sms_access_key_id,
),
aliyun_sms_access_key_secret=_config_text(
values,
"aliyun_sms_access_key_secret",
settings.aliyun_sms_access_key_secret,
aliyun_sms_access_key_secret=SecretService.decrypt(
_config_text(values, "aliyun_sms_access_key_secret", settings.aliyun_sms_access_key_secret)
),
aliyun_sms_sign_name=_config_text(values, "aliyun_sms_sign_name", settings.aliyun_sms_sign_name),
aliyun_sms_template_code=_config_text(