1. はじめに: さあ、コードに触れる時間です
こんにちは！前編ではGitHub Webhook自動デプロイシステムを直接構築する理由と必要な準備物について、2編では全体のアーキテクチャとFastAPIウェブフックサービスの主要な役割について詳しく見てきました。今度は頭の中で描いた絵を実際のコードに移してみる時間です。
前回のエピソードをまだ見ていない方は、以下のリンクで確認できます。
GitHub Webhookを活用した自分だけの自動デプロイシステム構築
1 - なぜ自分で実装するのか？ 2 -全体アーキテクチャとプロセス設計
今回は、ステージングサーバーの基本的な環境設定を完了し、GitHub Webhookリクエストを安全に受信し検証できるFastAPIウェブフックサーバーの基礎骨格を一緒に作成していきます。漠然と感じていた自動デプロイシステムが一段と現実に近づく時間です！
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ウェブフックサーバーの基本構造作成
今度はウェブフックリクエストを受け入れるFastAPIアプリケーションの基本骨組みを作成してみましょう。~/projects/webhook_serverディレクトリの中にmain.pyファイルを作成し、以下のコードを記述します。
3.1. main.pyファイル作成とFastAPIアプリ初期化
私は説明のためにmain.pyにすべてのロジックを一度に書きましたが、皆さんは実際のプロジェクトで管理のために各ユーティリティ関数ごとに個別のファイルでモジュール化し、importするなど、自分のスタイルに合わせて見やすく作成してください。以下は記述例です。
# ~/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サーバーは稼働中です！"}
# GitHub Webhookを受信するエンドポイント
@app.post("/webhook")
async def github_webhook(request: Request, background_tasks: BackgroundTasks):
logger.info("Webhookリクエストを受信しました。")
# 1. GitHub Webhook Secret検証 (最初に実行すべきセキュリティステップ)
if GITHUB_WEBHOOK_SECRET:
signature = request.headers.get("X-Hub-Signature-256")
if not signature:
logger.warning("Signatureヘッダーがありません。中止します。")
raise HTTPException(status_code=401, detail="X-Hub-Signature-256ヘッダーがありません")
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("無効な署名。中止します。")
raise HTTPException(status_code=401, detail="無効な署名")
else:
logger.info("署名が正常に検証されました。")
else:
logger.warning("GITHUB_WEBHOOK_SECRETが設定されていません。署名検証をスキップします（本番では推奨されません）。")
# 2. HTTPボディのパース (Payload)
try:
payload = await request.json()
event_type = request.headers.get("X-GitHub-Event")
# GitHub Webhookペイロードからリポジトリ名を抽出します。
# 'full_name'の代わりに'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_type}, リポジトリ: {repo_name}, リファレンス: {ref}, プッシャー: {pusher_name}")
# 実際のデプロイロジックをバックグラウンドで実行
background_tasks.add_task(handle_deploy, repo_name)
except Exception as e:
logger.error(f"Webhookペイロードのパースまたはリクエスト処理中にエラーが発生しました: {e}")
raise HTTPException(status_code=400, detail=f"リクエスト処理中にエラーが発生しました: {e}")
# GitHubに即応答を返す
return {"message": "Webhookが受信され、バックグラウンドで処理が開始されました！"}
# 実際のデプロイロジックを処理する関数
def handle_deploy(repo_name: str):
# 'deployer'という名前のリポジトリはウェブフックサーバー自体を更新する場合なのでスキップ
if repo_name == "deployer":
logger.info("⚙️ Deployerの自己更新をスキップしました。")
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 (project name)オプションで使用できます。
# Dockerの特性上、-pオプションには大文字やハイフンを使えないことに注意してください。
p_name = {
"sample_project1": "p_name_1",
"sample_project2": "p_name_2",
"sample_project3": "p_name_3",
}
# 現在ウェブフックリクエストが来たリポジトリのローカルパスを探します。
repo_path: Optional[str] = repo_paths.get(repo_name)
# 定義されていないリポジトリの場合、警告ログを残して終了します。
if not repo_path:
logger.warning(f"🚫 不明なリポジトリ: {repo_name}")
return
# Pathオブジェクトに変換してファイルシステムパスを簡単に扱うことができます。
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"{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"{repo_name}のGit pullが成功しました。")
except subprocess.CalledProcessError as e:
logger.error(f"{repo_name}のGit pullが失敗しました: {e.stderr}")
return # デプロイ失敗時にここで終了
logger.info(f"{repo_name}の変更を確認中...")
# 変更比較: Dockerイメージ再ビルドが必要か判断します。
changed = should_rebuild(repo_path)
try:
if changed:
logger.info(f"🚀 変更が検出されました。{docker_compose_file}で{repo_name}をビルドしデプロイします...")
# 変更があればDocker Composeを--buildオプション付きで実行し、イメージを再ビルドします。
subprocess.run(["docker", "compose", "-p", project_name, "-f", compose_file, "up", "-d", "--build"], check=True, capture_output=True, text=True)
logger.info(f"{repo_name}のDocker composeビルドとアップが成功しました。")
else:
logger.info(f"✨ 特に重要な変更はありません。{docker_compose_file}を使って{repo_name}を再ビルドせずにデプロイします...")
# 変更がなければ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"{repo_name}のDocker composeアップが成功しました。")
except subprocess.CalledProcessError as e:
logger.error(f"{repo_name}のDocker composeが失敗しました: {e.stderr}")
return # デプロイ失敗時にここで終了
logger.info(f"{repo_name}のデプロイタスクが完了しました。")
# Dockerイメージ再ビルドが必要か判断する関数
def should_rebuild(repo_path: Path) -> bool:
# このファイルが変更された場合、Dockerイメージを再ビルドする必要があります。
trigger_files = [
"requirements.txt", "Dockerfile", "Dockerfile.dev", "Dockerfile.prod",
".dockerignore", ".env" # .envファイル変更も再ビルドトリガーとして含む
]
logger.info(f"{repo_path}のgit diffを確認中...")
# 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"Gitで変更されたファイル: {changed_files_in_git}")
except subprocess.CalledProcessError as e:
logger.error(f"{repo_path}のgit diffに失敗しました: {e.stderr}")
# git diff失敗時に安全に再ビルドを試みることが推奨されます。
return True
# 変更されたファイルのリストにトリガーファイルが含まれているか確認します。
if any(f in changed_files_in_git for f in trigger_files):
logger.info("git diffでトリガーファイルの変更が検出されました。再ビルドが必要です。")
return True
# 手動の再ビルドトリガーファイル(REBUILD_TRIGGER)が存在するか確認します。
# このファイルがあれば、手動で再ビルドを強制するものと見なします。ファイルを削除します。
rebuild_trigger_file = repo_path / "REBUILD_TRIGGER"
if rebuild_trigger_file.exists():
try:
rebuild_trigger_file.unlink() # ファイル削除
logger.info("REBUILD_TRIGGERファイルが検出されました。再ビルドが必要で、トリガーファイルは削除されました。")
return True
except Exception as e:
logger.error(f"REBUILD_TRIGGERファイル削除中にエラーが発生しました: {e}")
# ファイル削除失敗時にも再ビルドを試みることが安全です。
return True
logger.info("再ビルドトリガーファイルが見つかりません。再ビルドは不要です。")
return False
3.2. /webhookエンドポイント定義
上記のコードで、app.post("/webhook")デコレータの下に定義された非同期関数github_webhookが、まさにGitHub Webhookリクエストを受信するエンドポイントです。この関数は、入ってくるPOSTリクエストを処理し、Requestオブジェクトを通じてヘッダーと本文（Payload）にアクセスできます。BackgroundTasksは、GitHubに即座に応答を送りつつ、デプロイなどの時間がかかる作業をバックグラウンドで実行するためのFastAPIの強力な機能です。
4. GitHub Webhook Secret検証ロジック実装
上記のコードで最も重要なセキュリティ要素は、GITHUB_WEBHOOK_SECRETを利用した検証ロジックです。
4.1. Secret値の重要性
GitHubウェブフックを設定する際、Secret値を設定できます。この値は、GitHubがウェブフックリクエストを送信する際、リクエスト本文とSecret値を組み合わせてハッシュ（Hash）値を生成し、これをX-Hub-Signature-256ヘッダーに含めて送ります。私たちのウェブフックサーバーは同様の方法でハッシュ値を生成し、GitHubが送ったハッシュ値と一致するか比較してリクエストがGitHubから来たものであるか、またデータが送信中に改ざんされていないか確認することができます。
4.2. X-Hub-Signature-256ヘッダーによる検証
コードでご覧のように、hmacおよびhashlibモジュールを使用してSecret検証を実施します。
- Secret設定: まず最初に、
~/projects/webhook_serverディレクトリに.envファイルを作成し、ここにGitHub Webhook設定時に使用するSecret値を記録します。このファイルは.gitignoreに追加し、Gitリポジトリに上がらないよう注意が必要です。このSecret値は後でGitHub RepositoryのWebhook設定で使用されることになります。
# ~/projects/webhook_server/.env
GITHUB_WEBHOOK_SECRET="ここに_GitHub_ウェブフック設定時に使用する_シークレット値を入力してください"
- FastAPIコード:
main.pyではpython-dotenvライブラリを使用してこの.envファイルをロードし、os.getenv()でGITHUB_WEBHOOK_SECRETの値を取得します。そして、ウェブフックリクエストがあるごとにRequest.headers.get("X-Hub-Signature-256")で署名値を取得し、hmac.new()関数を使って自ら想定する署名値を計算し、比較します。
この検証ロジックはウェブフックエンドポイントに対する外部攻撃を防ぐための必須の盾なので、絶対に省略しないでください！
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_server.logファイルに情報が記録され、同時にターミナルにも出力されます。のちにデプロイロジックが複雑になった際、このロギング機能を通じてどの段階で問題が発生したのかを容易に把握できます。
6. Uvicornでウェブフックサーバーを実行し、ローカルテスト
今度は作成したFastAPIウェブフックサーバーが正しく動作するかローカルで簡単にテストしてみましょう。ステージングサーバーの~/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ウェブフックサーバーの基本的骨格を完成させました。これでウェブフックサーバーが外部リクエストを安全に受け取る準備が整いました。
次回の4編では、今日作った基礎の上に実際のデプロイロジック（Git Pull、Dockerイメージの再ビルド判断、Docker Composeの実行）をsubprocessモジュールを活用して実装し、このFastAPIウェブフックサーバーがサーバー再起動時にも自動で実行されるようSystemdサービスに登録する方法を詳しく扱います。お楽しみに！
コメントはありません。