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。

FastAPI Webhook 伺服器架構圖

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 倉庫的 Webhook 設置中使用。
# ~/projects/webhook_server/.env
GITHUB_WEBHOOK_SECRET="請在此輸入 GitHub Webhook 設置時使用的 Secret 值"
  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 命令發送測試請求。

```bash

將伺服器的 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），並詳細討論如何將此 FastAPI Webhook 伺服器註冊為Systemd 服務，以便於伺服器重啟時自動運行。敬請期待！