marco
4b98d405d7
- app_config 加 get/set_app_version(独立 key,复用表不进 CONFIG_DEFS) - GET /api/v1/platform/app-version 公开读 + POST /internal/app-version 内部写(X-Internal-Secret) - tests/test_app_version.py 5 端到端测试 Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
59 lines
2.2 KiB
Python
59 lines
2.2 KiB
Python
"""最新 App 版本写入端点(发布流程 → app-server)。
|
|
|
|
发车流程出 APK 后,把版本号 / 下载链接 / sha256 等 POST 到这里,落 app_config(key=latest_app_version)。
|
|
客户端再 GET /api/v1/platform/app-version 读取做 OTA 检查更新。**不是给客户端的接口**:
|
|
不走用户 JWT,靠 server 间共享密钥头 `X-Internal-Secret` 校验(== settings.INTERNAL_API_SECRET)。
|
|
密钥未配置(默认空)时直接 503,避免裸奔的写端点。也是应急改版本信息(紧急下线/改 apk_url)的入口。
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import hmac
|
|
import logging
|
|
from typing import Annotated
|
|
|
|
from fastapi import APIRouter, Header, HTTPException, status
|
|
|
|
from app.api.deps import DbSession
|
|
from app.core.config import settings
|
|
from app.repositories import app_config
|
|
from app.schemas.platform import AppVersionOut, AppVersionWriteIn
|
|
|
|
logger = logging.getLogger("shagua.internal.app_version")
|
|
|
|
router = APIRouter(prefix="/internal", tags=["internal"])
|
|
|
|
|
|
def _check_secret(x_internal_secret: str | None) -> None:
|
|
"""共享密钥校验。未配置 → 503(挡住裸奔写端点);不匹配 → 401(常量时间比较)。"""
|
|
configured = settings.INTERNAL_API_SECRET
|
|
if not configured:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
|
|
detail="internal api not configured",
|
|
)
|
|
if not x_internal_secret or not hmac.compare_digest(x_internal_secret, configured):
|
|
raise HTTPException(
|
|
status_code=status.HTTP_401_UNAUTHORIZED,
|
|
detail="invalid internal secret",
|
|
)
|
|
|
|
|
|
@router.post(
|
|
"/app-version",
|
|
response_model=AppVersionOut,
|
|
summary="写最新 App 版本(发布流程→app-server,落 app_config)",
|
|
)
|
|
def write_app_version(
|
|
payload: AppVersionWriteIn,
|
|
db: DbSession,
|
|
x_internal_secret: Annotated[str | None, Header()] = None,
|
|
) -> AppVersionOut:
|
|
_check_secret(x_internal_secret)
|
|
data = payload.model_dump()
|
|
app_config.set_app_version(db, data)
|
|
logger.info(
|
|
"app_version set code=%d name=%s url=%s",
|
|
payload.latest_version_code, payload.latest_version_name, payload.apk_url,
|
|
)
|
|
return AppVersionOut(**data)
|