admin/app/routers/deploy.py

import hashlib
import hmac
import os
import subprocess
from datetime import datetime
from pathlib import Path
from typing import Any
from urllib.parse import parse_qs

from fastapi import APIRouter, BackgroundTasks, 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


def append_deploy_log(content: str) -> None:
    log_path = Path(deploy_settings.log_path)
    log_path.parent.mkdir(parents=True, exist_ok=True)
    with log_path.open("a", encoding="utf-8") as file:
        file.write(content)
        if not content.endswith("\n"):
            file.write("\n")


def build_deploy_env() -> dict[str, str]:
    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,
            "DEPLOY_NODE_BIN_DIR": deploy_settings.node_bin_dir,
        }
    )
    return env


def run_deploy_task(lock_fd: int) -> None:
    started_at = datetime.utcnow().replace(microsecond=0).isoformat()
    append_deploy_log(f"\n[{started_at}] deploy started")
    try:
        result = subprocess.run(
            [deploy_settings.script_path],
            cwd=deploy_settings.project_dir,
            env=build_deploy_env(),
            capture_output=True,
            text=True,
            timeout=deploy_settings.command_timeout,
            check=False,
        )
        finished_at = datetime.utcnow().replace(microsecond=0).isoformat()
        append_deploy_log(
            "\n".join(
                [
                    f"[{finished_at}] deploy finished returncode={result.returncode}",
                    "--- stdout ---",
                    output_tail(result.stdout, 20000),
                    "--- stderr ---",
                    output_tail(result.stderr, 20000),
                    "--- end ---",
                ]
            )
        )
    except subprocess.TimeoutExpired:
        finished_at = datetime.utcnow().replace(microsecond=0).isoformat()
        append_deploy_log(f"[{finished_at}] deploy timeout after {deploy_settings.command_timeout}s")
    except Exception as exc:
        finished_at = datetime.utcnow().replace(microsecond=0).isoformat()
        append_deploy_log(f"[{finished_at}] deploy error: {exc}")
    finally:
        release_lock(lock_fd)


@router.post("/webhook", status_code=status.HTTP_202_ACCEPTED)
async def deploy_webhook(request: Request, background_tasks: BackgroundTasks) -> 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()
    background_tasks.add_task(run_deploy_task, lock_fd)

    return {
        "status": "accepted",
        "message": "已接收推送事件，前端构建已在后台开始",
        "log_path": deploy_settings.log_path,
    }