chore: encrypt_secret.py

common/pyproject.toml

@@ -14,6 +14,7 @@ dependencies = [
     "bcrypt>=4.0,<4.1",
     "aiofiles>=25.1.0",
     "aioboto3>=15.5.0",
+    "cryptography>=49.0.0",
 ]
 
 [build-system]

common/src/common/config.py

@@ -17,13 +17,43 @@ Rules for adding a new variable:
 
 from __future__ import annotations
 
+import base64
+import os
 from functools import lru_cache
-from typing import Annotated
+from typing import Annotated, Any
+from venv import logger
 
-from pydantic import Field, field_validator
+from cryptography.hazmat.primitives.ciphers.aead import AESGCM
+from pydantic import Field, field_validator, model_validator
 from pydantic_settings import BaseSettings, NoDecode, SettingsConfigDict
 
 
+# ── AES 解密核心函数 ────────────────────────────────────────────────────────
+def _decrypt_value(cipher_text: str) -> str:
+    """解密 ENC(...) 格式的字符串。密钥从环境变量 APP_CONFIG_SECRET_KEY 获取。"""
+    if not (cipher_text.startswith("ENC(") and cipher_text.endswith(")")):
+        return cipher_text
+
+    secret_key = os.getenv("APP_CONFIG_SECRET_KEY")
+    if not secret_key:
+        raise RuntimeError(
+            "致命错误: 检测到配置项包含 ENC(...) 密文，但系统环境变量 "
+            "APP_CONFIG_SECRET_KEY 未设置！"
+        )
+
+    raw_payload = cipher_text[4:-1]
+    try:
+        data = base64.b64decode(raw_payload)
+        nonce, ciphertext = data[:12], data[12:]
+        # 将传入的 key 补全或截断为 32 字节 (AES-256)
+        key_bytes = secret_key.encode("utf-8").ljust(32, b"\0")[:32]
+        cipher = AESGCM(key_bytes)
+        decrypted = cipher.decrypt(nonce, ciphertext, None)
+        return decrypted.decode("utf-8")
+    except Exception as e:
+        raise ValueError(f"配置项解密失败，请检查密钥或密文正确性: {e}") from e
+
+
 class Settings(BaseSettings):
     # ── database ──────────────────────────────────────────────────
     database_url: str = Field(
@@ -95,7 +125,6 @@ class Settings(BaseSettings):
     @field_validator("audit_excluded_paths", mode="before")
     @classmethod
     def _split_audit_paths(cls, v):
-        # env 进来是 "a,b,c"；代码里直接传 list 也行
         if isinstance(v, str):
             return [s.strip() for s in v.split(",") if s.strip()]
         return v
@@ -223,13 +252,21 @@ class Settings(BaseSettings):
     )
 
     # ── readiness probes ──────────────────────────────────────────
-    readiness_targets: str = Field(
-        default="",
-        description=(
-            "Comma-separated host:port list checked by /health/ready. "
-            "Empty disables the check."
-        ),
-    )
+    readiness_targets: str = Field(default="")
+
+    # ── 全局密文拦截器 ───────────────────────────────────────────
+    @model_validator(mode="before")
+    @classmethod
+    def _decrypt_encrypted_fields(cls, values: dict[str, Any]) -> dict[str, Any]:
+        """在 Pydantic 赋值前自动扫描所有 str 类型的环境变量，解密 ENC(...)"""
+        if not isinstance(values, dict):
+            return values
+
+        for key, val in values.items():
+            if isinstance(val, str) and val.startswith("ENC("):
+                values[key] = _decrypt_value(val)
+                print(values[key])
+        return values
 
     model_config = SettingsConfigDict(
         env_file=".env",
@@ -247,5 +284,4 @@ def get_settings() -> Settings:
 
 settings: Settings = get_settings()
 
-
 __all__ = ["Settings", "get_settings", "settings"]