feat(deploy): 添加 Gitea 自动部署 Webhook 及配套工具
新增 `/deploy/webhook` 接口,支持 Gitea HMAC 签名校验、指定分支部署、构建并发锁与超时控制 添加前端项目部署脚本 `scripts/deploy_front.sh`,负责拉取代码、安装依赖并执行构建 新增部署相关配置项并更新 `.env.example` 示例配置 更新项目文档说明自动部署的使用约定 补充 `.gitignore` 规则忽略证书和 PEM 格式文件
This commit is contained in:
@@ -0,0 +1,171 @@
|
||||
import hashlib
|
||||
import hmac
|
||||
import os
|
||||
import subprocess
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
from urllib.parse import parse_qs
|
||||
|
||||
from fastapi import APIRouter, HTTPException, Request, status
|
||||
|
||||
from app.config import deploy_settings
|
||||
|
||||
|
||||
router = APIRouter(prefix="/deploy", tags=["deploy"])
|
||||
|
||||
|
||||
def get_request_token(request: Request) -> str:
|
||||
return (
|
||||
request.headers.get("x-deploy-token")
|
||||
or request.headers.get("x-gitlab-token")
|
||||
or request.query_params.get("token")
|
||||
or ""
|
||||
)
|
||||
|
||||
|
||||
def signature_matches(secret: str, body: bytes, signature: str | None) -> bool:
|
||||
if not signature:
|
||||
return False
|
||||
|
||||
expected = hmac.new(secret.encode("utf-8"), body, hashlib.sha256).hexdigest()
|
||||
normalized = signature.strip()
|
||||
if normalized.startswith("sha256="):
|
||||
normalized = normalized.removeprefix("sha256=")
|
||||
return hmac.compare_digest(normalized, expected)
|
||||
|
||||
|
||||
def verify_webhook_secret(request: Request, body: bytes) -> None:
|
||||
if not deploy_settings.webhook_secret:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
|
||||
detail="自动部署密钥未配置,请先设置 DEPLOY_WEBHOOK_SECRET",
|
||||
)
|
||||
|
||||
token = get_request_token(request)
|
||||
if token and hmac.compare_digest(token, deploy_settings.webhook_secret):
|
||||
return
|
||||
|
||||
signatures = (
|
||||
request.headers.get("x-hub-signature-256"),
|
||||
request.headers.get("x-gitea-signature"),
|
||||
request.headers.get("x-gogs-signature"),
|
||||
)
|
||||
if any(signature_matches(deploy_settings.webhook_secret, body, value) for value in signatures):
|
||||
return
|
||||
|
||||
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="自动部署密钥不正确")
|
||||
|
||||
|
||||
def parse_payload(body: bytes, content_type: str) -> dict[str, Any]:
|
||||
if not body:
|
||||
return {}
|
||||
|
||||
payload_body = body
|
||||
if "application/x-www-form-urlencoded" in content_type.lower():
|
||||
form = parse_qs(body.decode("utf-8", errors="replace"))
|
||||
payload_values = form.get("payload")
|
||||
if not payload_values:
|
||||
return {}
|
||||
payload_body = payload_values[0].encode("utf-8")
|
||||
|
||||
try:
|
||||
import json
|
||||
|
||||
value = json.loads(payload_body)
|
||||
except ValueError:
|
||||
return {}
|
||||
|
||||
return value if isinstance(value, dict) else {}
|
||||
|
||||
|
||||
def should_deploy(payload: dict[str, Any]) -> bool:
|
||||
if not deploy_settings.branch:
|
||||
return True
|
||||
|
||||
ref = str(payload.get("ref") or "")
|
||||
branch = str(payload.get("branch") or "")
|
||||
expected_ref = f"refs/heads/{deploy_settings.branch}"
|
||||
return ref == expected_ref or branch == deploy_settings.branch
|
||||
|
||||
|
||||
def output_tail(value: str, max_length: int = 4000) -> str:
|
||||
value = value.strip()
|
||||
if len(value) <= max_length:
|
||||
return value
|
||||
return value[-max_length:]
|
||||
|
||||
|
||||
def acquire_lock() -> int:
|
||||
try:
|
||||
return os.open(deploy_settings.lock_path, os.O_CREAT | os.O_EXCL | os.O_WRONLY)
|
||||
except FileExistsError as exc:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_409_CONFLICT,
|
||||
detail="前端正在构建中,请稍后再试",
|
||||
) from exc
|
||||
|
||||
|
||||
def release_lock(fd: int) -> None:
|
||||
os.close(fd)
|
||||
try:
|
||||
os.unlink(deploy_settings.lock_path)
|
||||
except FileNotFoundError:
|
||||
pass
|
||||
|
||||
|
||||
@router.post("/webhook")
|
||||
async def deploy_webhook(request: Request) -> dict[str, Any]:
|
||||
body = await request.body()
|
||||
verify_webhook_secret(request, body)
|
||||
|
||||
payload = parse_payload(body, request.headers.get("content-type", ""))
|
||||
if not should_deploy(payload):
|
||||
return {
|
||||
"status": "ignored",
|
||||
"message": f"当前推送不是部署分支 {deploy_settings.branch}",
|
||||
"ref": payload.get("ref"),
|
||||
}
|
||||
|
||||
script_path = Path(deploy_settings.script_path)
|
||||
if not script_path.exists():
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
|
||||
detail=f"部署脚本不存在:{script_path}",
|
||||
)
|
||||
|
||||
lock_fd = acquire_lock()
|
||||
try:
|
||||
env = os.environ.copy()
|
||||
env.update(
|
||||
{
|
||||
"DEPLOY_PROJECT_DIR": deploy_settings.project_dir,
|
||||
"DEPLOY_FRONT_DIR": deploy_settings.front_dir,
|
||||
"DEPLOY_BRANCH": deploy_settings.branch,
|
||||
}
|
||||
)
|
||||
result = subprocess.run(
|
||||
[str(script_path)],
|
||||
cwd=deploy_settings.project_dir,
|
||||
env=env,
|
||||
capture_output=True,
|
||||
text=True,
|
||||
timeout=deploy_settings.command_timeout,
|
||||
check=False,
|
||||
)
|
||||
except subprocess.TimeoutExpired as exc:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_504_GATEWAY_TIMEOUT,
|
||||
detail=f"部署命令超过 {deploy_settings.command_timeout} 秒未完成",
|
||||
) from exc
|
||||
finally:
|
||||
release_lock(lock_fd)
|
||||
|
||||
response = {
|
||||
"status": "success" if result.returncode == 0 else "failed",
|
||||
"returncode": result.returncode,
|
||||
"stdout": output_tail(result.stdout),
|
||||
"stderr": output_tail(result.stderr),
|
||||
}
|
||||
if result.returncode != 0:
|
||||
raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=response)
|
||||
return response
|
||||
Reference in New Issue
Block a user