feat/ 导出开发容器内的数据用于评估

api/app/features/evaluation/internal_auth.py

@@ -0,0 +1,61 @@
+"""Internal evaluation API：共享密钥鉴权，不依赖终端用户 JWT。"""
+
+from typing import Annotated
+
+from fastapi import Depends, Header, HTTPException, status
+
+from app.core.config import settings
+from app.core.logging import get_logger
+
+logger = get_logger(__name__)
+
+INTERNAL_HEADER = "X-Internal-Eval-Key"
+
+
+class InternalEvalPrincipal:
+    """已通过内部密钥校验的调用方（占位，便于后续扩展多密钥/审计）。"""
+
+    def __init__(self, *, key_id: str = "default") -> None:
+        self.key_id = key_id
+
+
+def require_internal_eval_enabled() -> None:
+    if not (settings.internal_eval_api_key or "").strip():
+        logger.warning("internal_eval_api_key 未配置，内部评测 API 拒绝访问")
+        raise HTTPException(
+            status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
+            detail="内部评测服务未启用（缺少 INTERNAL_EVAL_API_KEY）",
+        )
+
+
+def verify_internal_eval_key(
+    *,
+    header_value: str | None = None,
+    query_value: str | None = None,
+) -> InternalEvalPrincipal:
+    """Header 或 query（供 EventSource 等无法带头场景）。"""
+    require_internal_eval_enabled()
+    expected = (settings.internal_eval_api_key or "").strip()
+    if not expected:
+        raise HTTPException(
+            status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
+            detail="内部评测服务未启用",
+        )
+    provided = (header_value or query_value or "").strip()
+    if not provided or provided != expected:
+        raise HTTPException(
+            status_code=status.HTTP_401_UNAUTHORIZED,
+            detail="无效的内部评测密钥",
+        )
+    return InternalEvalPrincipal()
+
+
+async def get_internal_eval_principal(
+    x_internal_eval_key: Annotated[str | None, Header(alias=INTERNAL_HEADER)] = None,
+) -> InternalEvalPrincipal:
+    return verify_internal_eval_key(header_value=x_internal_eval_key)
+
+
+InternalEvalAuth = Annotated[
+    InternalEvalPrincipal, Depends(get_internal_eval_principal)
+]