利用GitHub Webhook构建我的自动部署系统 ③ 阶段服务器环境设置与FastAPI Webhook服务器基础构建

1. 引言：是时候动手写代码了

---

---

大家好！在第一部分中，我们讨论了为什么要自己构建GitHub Webhook自动部署系统及其所需的准备工作；第二部分则详细探讨了整体架构和FastAPI Webhook服务的核心作用。现在是时候将我们脑中构思的图景转换为实际代码了。

如果您还没有看到之前的部分，可以通过以下链接查看。

利用GitHub Webhook构建我的自动部署系统

1 - 为什么要自己实现？ 2 - 整体架构及过程设计

在这第三部分中，我们将完成阶段服务器的基本环境设置，并共同构建一个能够安全接收和验证GitHub Webhook请求的FastAPI Webhook服务器基础框架。看似遥不可及的自动部署系统正在逐渐变为现实！

2. 阶段服务器初始环境设置

首先要做的是准备好将要进行部署的阶段服务器环境，以便我们能够执行代码。请通过SSH连接到服务器并按照以下步骤进行。

2.1. 准备Python开发环境

在服务器上设置Python虚拟环境是一个良好的习惯，可以防止项目之间的依赖冲突并保持干净的开发环境。在venv环境中使用pip安装所需的包。

# 安装FastAPI、Uvicorn和用于读取.env文件的python-dotenv
pip install fastapi uvicorn python-dotenv

2.2. 验证Git、Docker和Docker Compose安装

正如上文提到的，Git、Docker和Docker Compose是自动部署的核心工具。我们假设在阅读这篇文章的读者们，这些工具已经安装在服务器上。

如果尚未安装，请先使用以下命令检查。

git --version
docker --version
docker compose version # 或 docker-compose --version（根据安装方式可能有所不同）

如果未安装，请参考每个工具的官方文档进行安装。特别是如果您需要Docker安装的详细指南，可以参考以下文章： Docker安装指南

注意: 安装Docker后，您需要将当前用户添加到Docker组，并重新登录以便没有sudo可以使用docker命令。

sudo usermod -aG docker $USER

授予Docker权限后，必须在服务器上注销并重新登录才能使更改生效。 请使用docker run hello-world命令验证是否正常运行，无需sudo。

3. 创建FastAPI Webhook服务器基本结构

---

---

现在让我们来创建接收Webhook请求的FastAPI应用程序的基础框架。在~/projects/webhook_server目录中创建一个main.py文件并编写以下代码。

3.1. 创建main.py文件并初始化FastAPI应用

为了说明方便，我在main.py中将所有逻辑一次性编写，但您可以根据您的项目进行管理，将每个实用函数模块化到单独的文件中并导入，尽量编写成合乎自己风格的代码。下面是编写的示例。

# ~/projects/webhook_server/main.py
import hmac
import hashlib
import os
import logging
import subprocess
from pathlib import Path
from typing import Optional

from fastapi import FastAPI, Request, HTTPException, BackgroundTasks
from dotenv import load_dotenv, dotenv_values

# 加载.env文件（用于使用环境变量）
load_dotenv()

# 设置日志
logging.basicConfig(
    level=logging.INFO, # 记录INFO级别以上的日志
    format='%(asctime)s - %(levelname)s - %(message)s',
    handlers=[
        logging.FileHandler("webhook_server.log"), # 保存日志到文件
        logging.StreamHandler() # 同时输出到控制台
    ]
)
logger = logging.getLogger(__name__)

app = FastAPI()

# 从环境变量中加载GitHub Webhook Secret
GITHUB_WEBHOOK_SECRET = os.getenv("GITHUB_WEBHOOK_SECRET")

# 测试用根端点：在实际运营中建议移除，以确保安全。
# 一旦暴露在互联网上，各种爬虫会随处活动，因此无需确保此类端点存在。
@app.get("/")
async def read_root():
    return {"message": "Webhook server is running!"}

# 接收GitHub Webhook的端点
@app.post("/webhook")
async def github_webhook(request: Request, background_tasks: BackgroundTasks):
    logger.info("Webhook request received.")

    # 1. 验证GitHub Webhook Secret（最先执行的安全步骤）
    if GITHUB_WEBHOOK_SECRET:
        signature = request.headers.get("X-Hub-Signature-256")
        if not signature:
            logger.warning("Signature header missing. Aborting.")
            raise HTTPException(status_code=401, detail="X-Hub-Signature-256 header missing")

        body = await request.body()
        expected_signature = "sha256=" + hmac.new(
            GITHUB_WEBHOOK_SECRET.encode('utf-8'),
            body,
            hashlib.sha256
        ).hexdigest()

        if not hmac.compare_digest(expected_signature, signature):
            logger.warning("Invalid signature. Aborting.")
            raise HTTPException(status_code=401, detail="Invalid signature")
        else:
            logger.info("Signature verified successfully.")
    else:
        logger.warning("GITHUB_WEBHOOK_SECRET is not set. Skipping signature verification (NOT RECOMMENDED FOR PRODUCTION).")


    # 2. 解析HTTP Body（Payload）
    try:
        payload = await request.json()
        event_type = request.headers.get("X-GitHub-Event")
        # 从GitHub Webhook Payload中提取仓库名称。
        # 使用'name'而非'full_name'。
        repo_name = payload.get("repository", {}).get("name", "unknown_repository")
        ref = payload.get("ref", "unknown_ref")
        pusher_name = payload.get("pusher", {}).get("name", "unknown_pusher")

        logger.info(f"Event: {event_type}, Repo: {repo_name}, Ref: {ref}, Pusher: {pusher_name}")

        # 在后台执行实际的部署逻辑
        background_tasks.add_task(handle_deploy, repo_name)

    except Exception as e:
        logger.error(f"Error parsing webhook payload or processing request: {e}")
        raise HTTPException(status_code=400, detail=f"Error processing request: {e}")

    # 立即返回响应给GitHub
    return {"message": "Webhook received and processing started in background!"}

# 处理实际部署逻辑的函数
def handle_deploy(repo_name: str):
    # 'deployer'这个名字的仓库是更新Webhook服务器本身的情况，因此跳过
    if repo_name == "deployer":
        logger.info("⚙️ Deployer self-update skipped.")
        return

    # 从环境变量中读取每个仓库的本地路径。
    # 这个字典应该映射您实际项目的名称和服务器内的路径。
    # 例如： SAMPLE_PROJECT_1_PATH环境变量中存储着该项目的服务器路径。
    repo_paths = {
        "sample_project1": os.getenv("SAMPLE_PROJECT_1_PATH"),
        "sample_project2": os.getenv("SAMPLE_PROJECT_2_PATH"),
        "sample_project3": os.getenv("SAMPLE_PROJECT_3_PATH"),
    }

    # 如果仓库名称和Docker Compose项目名称不同，可以重新映射项目名称。
    # 这样设置后，可以使用Docker Compose的-p（项目名称）选项。
    # 注意：因Docker的特性，-p选项中不能使用大写或连字符。
    p_name = {
        "sample_project1": "p_name_1",
        "sample_project2": "p_name_2",
        "sample_project3": "p_name_3",
    }

    # 找到当前Webhook请求的仓库的本地路径
    repo_path: Optional[str] = repo_paths.get(repo_name)

    # 如果是未定义的仓库，就记录警告日志并退出。
    if not repo_path:
        logger.warning(f"🚫 Unknown repository: {repo_name}")
        return

    # 将路径对象转换，方便操作文件系统路径。
    repo_path = Path(repo_path)

    # 从对应仓库的.env文件中加载DEBUG值和COLOR值。
    # 这可以用于区分项目间的开发/生产环境，或动态设置Docker Compose项目名称。
    env_path = repo_path / ".env"
    env_vars = dotenv_values(env_path) if env_path.exists() else {}
    is_debug = env_vars.get("DEBUG", "false").lower() == "true" # 将"true"字符串转换为布尔值
    deploy_color = env_vars.get("COLOR", "red") # 默认值为"red"
    project_name = str(deploy_color) + p_name.get(repo_name, repo_name) # Docker Compose项目名称

    # 根据DEBUG值决定使用的Docker Compose文件。
    docker_compose_file = "docker-compose.dev.yml" if is_debug else "docker-compose.prod.yml"

    # 定义Docker Compose文件路径。
    compose_file = repo_path / docker_compose_file

    logger.info(f"Pulling latest code for {repo_name} in {repo_path}...")
    # 执行Git Pull：从对应仓库路径获取最新代码。
    # check=True选项在命令执行失败时会引发subprocess.CalledProcessError，以方便异常处理。
    try:
        subprocess.run(["git", "-C", repo_path, "pull"], check=True, capture_output=True, text=True)
        logger.info(f"Git pull successful for {repo_name}.")
    except subprocess.CalledProcessError as e:
        logger.error(f"Git pull failed for {repo_name}: {e.stderr}")
        return # 部署失败则退出

    logger.info(f"Checking for changes in {repo_name} to decide rebuild...")
    # 比较更改：判断Docker镜像是否需要重新构建。
    changed = should_rebuild(repo_path)

    try:
        if changed:
            logger.info(f"🚀 Changes detected. Building and deploying {repo_name} with {docker_compose_file}...")
            # 如果有更改，使用--build选项执行Docker Compose以重新构建镜像。
            subprocess.run(["docker", "compose", "-p", project_name, "-f", compose_file, "up", "-d", "--build"], check=True, capture_output=True, text=True)
            logger.info(f"Docker compose build and up successful for {repo_name}.")
        else:
            logger.info(f"✨ No significant changes. Deploying {repo_name} without rebuilding images using {docker_compose_file}...")
            # 如果没有更改，则执行Docker Compose而不使用--build选项。
            subprocess.run(["docker", "compose", "-p", project_name, "-f", compose_file, "up", "-d"], check=True, capture_output=True, text=True)
            logger.info(f"Docker compose up successful for {repo_name}.")
    except subprocess.CalledProcessError as e:
        logger.error(f"Docker compose failed for {repo_name}: {e.stderr}")
        return # 部署失败则退出

    logger.info(f"✅ Deployment task for {repo_name} finished.")

# 判断是否需要重新构建Docker镜像的函数
def should_rebuild(repo_path: Path) -> bool:
    # 这些文件的更改将触发Docker镜像的重新构建。
    trigger_files = [
        "requirements.txt", "Dockerfile", "Dockerfile.dev", "Dockerfile.prod",
        ".dockerignore", ".env" # 也将.env文件更改视为触发重新构建的条件
    ]

    logger.info(f"Checking git diff for {repo_path}...")
    # 使用Git diff获取HEAD~1（上一个提交）和当前HEAD之间更改的文件列表。
    # --name-only选项仅获取更改的文件名。
    try:
        result = subprocess.run(
            ["git", "diff", "--name-only", "HEAD~1"],
            cwd=repo_path,
            capture_output=True, text=True, check=True
        )
        changed_files_in_git = result.stdout.strip().splitlines()
        logger.info(f"Changed files in git: {changed_files_in_git}")
    except subprocess.CalledProcessError as e:
        logger.error(f"Git diff failed for {repo_path}: {e.stderr}")
        # 如果git diff失败，建议安全地尝试重新构建。
        return True

    # 检查更改的文件列表是否包含触发文件。
    if any(f in changed_files_in_git for f in trigger_files):
        logger.info("Detected trigger file changes via git diff. Rebuild required.")
        return True

    # 检查是否存在手动重建触发文件（REBUILD_TRIGGER）。
    # 如果存在这个文件，则认为是强制手动重建，并删除该文件。
    rebuild_trigger_file = repo_path / "REBUILD_TRIGGER"
    if rebuild_trigger_file.exists():
        try:
            rebuild_trigger_file.unlink() # 删除文件
            logger.info("Detected REBUILD_TRIGGER file. Rebuild required and trigger file removed.")
            return True
        except Exception as e:
            logger.error(f"Error removing REBUILD_TRIGGER file: {e}")
            # 删除文件失败时也安全地尝试重建。
            return True

    logger.info("No rebuild trigger files found. No rebuild required.")
    return False

3.2. 定义/webhook端点

在上面的代码中，app.post("/webhook")装饰器下定义的异步函数github_webhook正是接收GitHub Webhook请求的端点。该函数处理传入的POST请求，并可以通过Request对象访问其头部和主体（Payload）。BackgroundTasks是FastAPI的强大功能，它能够在立即响应GitHub的同时，后台执行时间较长的任务，如部署。

4. 实现GitHub Webhook Secret验证逻辑

在上面的代码中，最重要的安全元素是利用GITHUB_WEBHOOK_SECRET进行的验证逻辑。

4.1. Secret值的重要性

在设置GitHub Webhook时，可以设置Secret值。该值用于GitHub在发送Webhook请求时，将请求主体与Secret值组合生成哈希（Hash）值，并通过X-Hub-Signature-256头部发送。我们的Webhook服务器将以相同的方式计算哈希值，并与GitHub发送的哈希值进行比较，以确保请求确实来自GitHub，并且数据在传输过程中未被篡改。

4.2. 利用X-Hub-Signature-256头进行验证

正如代码所示，使用hmac和hashlib模块执行Secret验证。

1. 设置Secret: 首先，在~/projects/webhook_server目录中创建.env文件，并在其中记录将在GitHub Webhook设置中使用的Secret值。注意此文件需添加到.gitignore中以避免被上传至Git仓库。这个Secret值将在之后用于GitHub Repository的Webhook设置。

# ~/projects/webhook_server/.env
GITHUB_WEBHOOK_SECRET="在此输入GitHub Webhook设置时使用的秘密值"

1. FastAPI代码: 在main.py中，使用python-dotenv库加载此.env文件，并通过os.getenv()获取GITHUB_WEBHOOK_SECRET的值。随后在Webhook请求到来时，通过Request.headers.get("X-Hub-Signature-256")获取签名值，利用hmac.new()函数计算期望的签名值并进行比较。

该验证逻辑是对Webhook端点的外部攻击的必要防护，绝不可跳过！

5. 设置日志系统

可以看到，在代码中使用logging模块进行了一些基本的日志设置。

logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s - %(message)s',
    handlers=[
        logging.FileHandler("webhook_server.log"), # 保存到webhook_server.log文件
        logging.StreamHandler() # 输出到控制台
    ]
)
logger = logging.getLogger(__name__)

这样设置后，每当有Webhook请求到来时，信息会被记录在webhook_server.log文件中，并同时输出到终端。随着部署逻辑的复杂化，此日志功能可以帮助您轻松判断哪个阶段出现了问题。

6. 使用Uvicorn运行Webhook服务器并进行本地测试

现在让我们在本地简单测试一下编写的FastAPI Webhook服务器是否正常工作。在阶段服务器的~/projects/webhook_server目录中，确保虚拟环境已激活，然后运行以下命令。

# 确保虚拟环境已激活（检查前面是否有(venv)）
# source venv/bin/activate # 如果未激活，请重新执行该命令

uvicorn main:app --host 0.0.0.0 --port 8000 --reload

--reload`选项非常便利，可以在开发中自动重启服务器以便代码更改生效。

请在浏览器中访问127.0.0.1:8000/docs，以确保API文档正确生成。如果能正常看到文档，则表明应用程序正在正常运行。如果知道服务器的IP地址（例如192.168.1.100），您可以在本地机器上使用以下curl命令发送测试请求。

# 将服务器的IP地址更改为'YOUR_SERVER_IP'。
# 'YOUR_WEBHOOK_SECRET'替换为.env文件中的实际值。

# 含有简单的虚拟JSON数据。
curl -X POST http://YOUR_SERVER_IP:8000/webhook \
     -H "Content-Type: application/json" \
     -H "X-GitHub-Event: push" \
     -H "X-Hub-Signature-256: sha256=$(echo -n '{"repository":{"name":"your-repo"},"ref":"refs/heads/main","pusher":{"name":"test-user"}}' | openssl dgst -sha256 -hmac 'YOUR_WEBHOOK_SECRET' | awk '{print $2}')" \
     -d '{"repository":{"name":"your-repo"},"ref":"refs/heads/main","pusher":{"name":"test-user"}}'

请检查终端是否输出了Webhook received and processing started in background!消息以及记录了日志，并且服务器日志文件(webhook_server.log)中也有记录。如果出现错误，您可以根据终端输出的错误消息进行调试。

7. 总结：下期预告

在本期中，我们构建了阶段服务器的环境，并完成了接收GitHub Webhook请求并验证Secret的FastAPI Webhook服务器的基础框架。现在Webhook服务器已经准备好安全接收外部请求。

在接下来的第四部分中，我们将基于今天创建的基础，利用subprocess模块实现实际部署逻辑（Git Pull、Docker镜像重新构建决策、Docker Compose执行），并详细介绍如何通过注册为Systemd服务使FastAPI Webhook服务器在服务器重启时自动启动。敬请期待！