admin/app/routers/deploy.py

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