feat(webhook): Task 3 — gitea_webhook.py router (ADR-059)

- 新增 Gitea Webhook Router: X-Gitea-Event/Signature/Delivery - 支援 pull_request / push / ping，移除 workflow_run - review_id prefix 改為 gt-pr-* / gt-push-* Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
2026-04-05 14:41:12 +08:00
parent 6777532534
commit b2c0148f2b
1 changed files with 498 additions and 0 deletions
--- a/apps/api/src/api/v1/gitea_webhook.py
+++ b/apps/api/src/api/v1/gitea_webhook.py
@@ -0,0 +1,498 @@
+"""
+AWOOOI API - Gitea Webhook Handler
+====================================
+ADR-059: GitHub → Gitea Webhook 遷移
+
+整合流程:
+1. Gitea Webhook (PR/Push) → AWOOOI API
+2. HMAC-SHA256 簽章驗證 (X-Gitea-Signature)
+3. 解析 PR diff / Push commits
+4. 呼叫 OpenClaw 進行 AI 代碼審查
+5. 儲存審查結果到 Redis
+6. 發送 Telegram 通知
+7. (可選) 建立 Approval 等待人工確認
+
+支援事件:
+- pull_request: PR 代碼審查
+- push: 主分支推送審查
+- ping: 連線測試
+
+安全要求:
+- HMAC 簽章驗證 (X-Gitea-Signature)
+- Webhook Secret 存放於 K8s Secret
+- 倉庫白名單驗證
+
+版本: v1.0
+最後修改: 2026-04-05 (台北時區)
+修改者: Claude Code (ADR-059 GitHub → Gitea 遷移)
+"""
+
+import hashlib
+import hmac
+import json
+import uuid
+from typing import Literal
+
+from fastapi import APIRouter, BackgroundTasks, Header, HTTPException, Request, status
+from pydantic import BaseModel
+
+from src.core.config import settings
+from src.core.logging import get_logger
+from src.services.gitea_webhook_service import get_gitea_webhook_service
+
+logger = get_logger("awoooi.gitea_webhook")
+
+router = APIRouter(prefix="/webhooks/gitea", tags=["Gitea Webhook"])
+
+# =============================================================================
+# Constants
+# =============================================================================
+
+# OpenClaw 配置 (使用 settings 中的 OPENCLAW_URL)
+OPENCLAW_URL = settings.OPENCLAW_URL
+
+# =============================================================================
+# Pydantic Models
+# =============================================================================
+
+class GiteaUser(BaseModel):
+    """Gitea 使用者"""
+    login: str
+    id: int
+    avatar_url: str | None = None
+
+
+class GiteaRepository(BaseModel):
+    """Gitea 倉庫"""
+    id: int
+    name: str
+    full_name: str
+    private: bool = False
+    html_url: str
+
+
+class GiteaPullRequest(BaseModel):
+    """Gitea PR 資訊"""
+    id: int
+    number: int
+    title: str
+    body: str | None = None
+    state: str  # open, closed
+    html_url: str
+    diff_url: str
+    user: GiteaUser
+    head: dict  # head branch info
+    base: dict  # base branch info
+    additions: int = 0
+    deletions: int = 0
+    changed_files: int = 0
+
+
+class GiteaCommit(BaseModel):
+    """Gitea Commit 資訊"""
+    id: str  # SHA
+    message: str
+    timestamp: str
+    url: str
+    author: dict
+    added: list[str] = []
+    removed: list[str] = []
+    modified: list[str] = []
+
+
+class GiteaWebhookPayload(BaseModel):
+    """Gitea Webhook Payload (通用)"""
+    action: str | None = None  # PR: opened, synchronize, etc.
+    repository: GiteaRepository
+    sender: GiteaUser
+    # PR 事件
+    pull_request: GiteaPullRequest | None = None
+    # Push 事件
+    ref: str | None = None  # refs/heads/main
+    before: str | None = None  # previous commit SHA
+    after: str | None = None  # current commit SHA
+    commits: list[GiteaCommit] | None = None
+    pusher: dict | None = None
+
+
+class GiteaWebhookResponse(BaseModel):
+    """Webhook 回應"""
+    status: Literal["accepted", "ignored", "error"]
+    message: str
+    event_type: str | None = None
+    review_id: str | None = None
+
+
+# =============================================================================
+# HMAC Signature Verification (CISO 安全要求)
+# =============================================================================
+
+class GiteaSignatureError(Exception):
+    """Gitea 簽章驗證失敗"""
+    pass
+
+
+async def verify_gitea_signature(
+    request: Request,
+    x_gitea_signature: str | None,
+) -> bool:
+    """
+    驗證 Gitea Webhook 請求的 HMAC-SHA256 簽章
+
+    - 簽章 Header: X-Gitea-Signature
+    - 簽章格式: sha256=<hex_digest>
+    - 使用 GITEA_WEBHOOK_SECRET 進行驗證
+
+    安全鐵律 (Fail-Closed):
+    - 生產環境: Secret 未設定 → 直接拒絕
+    - 開發環境: 可跳過驗證 (僅供本地測試)
+    """
+    if not settings.GITEA_WEBHOOK_SECRET:
+        if settings.ENVIRONMENT == "prod":
+            logger.critical(
+                "gitea_webhook_secret_missing_in_production",
+                environment=settings.ENVIRONMENT,
+                message="CRITICAL: GITEA_WEBHOOK_SECRET missing in production!",
+            )
+            raise GiteaSignatureError(
+                "Critical: GITEA_WEBHOOK_SECRET missing in production environment"
+            )
+
+        logger.warning(
+            "gitea_signature_verification_skipped_dev_only",
+            environment=settings.ENVIRONMENT,
+            reason="GITEA_WEBHOOK_SECRET not configured (dev mode only)",
+        )
+        return True
+
+    if not x_gitea_signature:
+        logger.warning("gitea_signature_missing")
+        raise GiteaSignatureError("Missing X-Gitea-Signature header")
+
+    if not x_gitea_signature.startswith("sha256="):
+        raise GiteaSignatureError("Invalid signature format (expected sha256=...)")
+
+    provided_signature = x_gitea_signature[7:]  # 移除 "sha256=" 前綴
+
+    body = await request.body()
+
+    expected_signature = hmac.new(
+        settings.GITEA_WEBHOOK_SECRET.encode(),
+        body,
+        hashlib.sha256,
+    ).hexdigest()
+
+    # 常數時間比較 (防止計時攻擊)
+    if not hmac.compare_digest(provided_signature, expected_signature):
+        logger.warning(
+            "gitea_signature_verification_failed",
+            provided=provided_signature[:16] + "...",
+            expected=expected_signature[:16] + "...",
+        )
+        raise GiteaSignatureError("Invalid signature")
+
+    logger.info("gitea_signature_verification_success")
+    return True
+
+
+def verify_gitea_allowed_repo(full_name: str) -> bool:
+    """
+    驗證倉庫是否在白名單中
+
+    Args:
+        full_name: 完整倉庫名稱 (owner/repo)
+
+    Returns:
+        bool: 是否允許
+    """
+    allowed_repos = settings.get_gitea_allowed_repos()
+
+    # 如果白名單為空，開發環境允許所有
+    if not allowed_repos:
+        if settings.ENVIRONMENT == "prod":
+            logger.warning(
+                "gitea_allowed_repos_empty_in_production",
+                repo=full_name,
+                message="No allowed repos configured in production",
+            )
+            return False
+        # 開發環境: 白名單空 = 允許所有
+        logger.debug(
+            "gitea_repo_allowed_dev_mode",
+            repo=full_name,
+            reason="Empty whitelist in dev mode",
+        )
+        return True
+
+    # 檢查是否在白名單中
+    is_allowed = full_name in allowed_repos
+    if not is_allowed:
+        logger.warning(
+            "gitea_repo_not_in_whitelist",
+            repo=full_name,
+            allowed_repos=allowed_repos,
+        )
+    return is_allowed
+
+
+# =============================================================================
+# Main Webhook Handler
+# =============================================================================
+
+@router.post(
+    "",
+    response_model=GiteaWebhookResponse,
+    status_code=status.HTTP_202_ACCEPTED,
+    summary="Gitea Webhook 接收端點",
+    description="接收 Gitea PR/Push 事件並觸發 AI 代碼審查",
+)
+async def handle_gitea_webhook(
+    request: Request,
+    background_tasks: BackgroundTasks,
+    x_gitea_event: str | None = Header(None, alias="X-Gitea-Event"),
+    x_gitea_delivery: str | None = Header(None, alias="X-Gitea-Delivery"),
+    x_gitea_signature: str | None = Header(None, alias="X-Gitea-Signature"),
+):
+    """
+    Gitea Webhook Handler
+
+    支援事件:
+    - pull_request (opened, synchronize, reopened)
+    - push (to default branch)
+
+    處理流程:
+    1. 驗證簽章
+    2. 驗證倉庫白名單
+    3. 解析事件類型
+    4. 背景執行 AI 審查 (委派給 GiteaWebhookService)
+    """
+    try:
+        # 1. 驗證 HMAC 簽章
+        try:
+            await verify_gitea_signature(request, x_gitea_signature)
+        except GiteaSignatureError as e:
+            logger.warning("gitea_webhook_signature_failed", error=str(e))
+            raise HTTPException(
+                status_code=status.HTTP_401_UNAUTHORIZED,
+                detail=str(e),
+            ) from e
+
+        # 2. 解析 Payload
+        body = await request.body()
+        payload_dict = json.loads(body)
+        payload = GiteaWebhookPayload(**payload_dict)
+
+        # 3. 驗證倉庫白名單
+        if not verify_gitea_allowed_repo(payload.repository.full_name):
+            return GiteaWebhookResponse(
+                status="ignored",
+                message=f"Repository {payload.repository.full_name} not in whitelist",
+                event_type=x_gitea_event,
+            )
+
+        # 4. 根據事件類型處理
+        logger.info(
+            "gitea_webhook_received",
+            gitea_event=x_gitea_event,
+            delivery_id=x_gitea_delivery,
+            repo=payload.repository.full_name,
+            sender=payload.sender.login,
+        )
+
+        # Pull Request 事件
+        if x_gitea_event == "pull_request":
+            return await handle_pull_request(
+                payload=payload,
+                background_tasks=background_tasks,
+                delivery_id=x_gitea_delivery,
+            )
+
+        # Push 事件
+        elif x_gitea_event == "push":
+            return await handle_push(
+                payload=payload,
+                background_tasks=background_tasks,
+                delivery_id=x_gitea_delivery,
+            )
+
+        # Ping 事件 (Gitea 測試連線)
+        elif x_gitea_event == "ping":
+            return GiteaWebhookResponse(
+                status="accepted",
+                message="Pong! Webhook configured successfully.",
+                event_type="ping",
+            )
+
+        # 其他事件 (忽略)
+        else:
+            return GiteaWebhookResponse(
+                status="ignored",
+                message=f"Event type '{x_gitea_event}' not supported",
+                event_type=x_gitea_event,
+            )
+
+    except HTTPException:
+        raise
+    except Exception as e:
+        logger.exception("gitea_webhook_processing_failed", error=str(e))
+        raise HTTPException(
+            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
+            detail="Internal error processing webhook",
+        ) from e
+
+
+# =============================================================================
+# Event Handlers (HTTP 層: 解析、驗證、回應 — 業務邏輯在 Service 層)
+# =============================================================================
+
+async def handle_pull_request(
+    payload: GiteaWebhookPayload,
+    background_tasks: BackgroundTasks,
+    delivery_id: str | None,  # noqa: ARG001 — reserved for idempotency (future use)
+) -> GiteaWebhookResponse:
+    """
+    處理 Pull Request 事件
+
+    支援 action:
+    - opened: 新建 PR
+    - synchronize: 推送新 commit 到 PR
+    - reopened: 重新開啟 PR
+    """
+    pr = payload.pull_request
+    if not pr:
+        return GiteaWebhookResponse(
+            status="error",
+            message="Missing pull_request data",
+            event_type="pull_request",
+        )
+
+    # 只處理需要審查的 action
+    supported_actions = {"opened", "synchronize", "reopened"}
+    if payload.action not in supported_actions:
+        return GiteaWebhookResponse(
+            status="ignored",
+            message=f"PR action '{payload.action}' not supported",
+            event_type="pull_request",
+        )
+
+    # 生成審查 ID
+    review_id = f"gitea-pr-{payload.repository.id}-{pr.number}-{uuid.uuid4().hex[:8]}"
+
+    # 背景執行審查 (委派給 Service)
+    service = get_gitea_webhook_service()
+    background_tasks.add_task(
+        service.review_pull_request,
+        repo=payload.repository,
+        pr=pr,
+        sender=payload.sender,
+        review_id=review_id,
+        action=payload.action,
+    )
+
+    logger.info(
+        "gitea_pr_review_scheduled",
+        review_id=review_id,
+        repo=payload.repository.full_name,
+        pr_number=pr.number,
+        pr_title=pr.title[:50],
+        action=payload.action,
+    )
+
+    return GiteaWebhookResponse(
+        status="accepted",
+        message=f"PR #{pr.number} review scheduled",
+        event_type="pull_request",
+        review_id=review_id,
+    )
+
+
+async def handle_push(
+    payload: GiteaWebhookPayload,
+    background_tasks: BackgroundTasks,
+    delivery_id: str | None,  # noqa: ARG001 — reserved for idempotency (future use)
+) -> GiteaWebhookResponse:
+    """
+    處理 Push 事件
+
+    只處理 default branch (main/master) 的 push
+    """
+    # 檢查是否推送到主分支
+    ref = payload.ref or ""
+    # 通常是 refs/heads/main 或 refs/heads/master
+    if not (ref.endswith("/main") or ref.endswith("/master")):
+        return GiteaWebhookResponse(
+            status="ignored",
+            message=f"Push to non-default branch: {ref}",
+            event_type="push",
+        )
+
+    commits = payload.commits or []
+    if not commits:
+        return GiteaWebhookResponse(
+            status="ignored",
+            message="No commits in push",
+            event_type="push",
+        )
+
+    # 生成審查 ID
+    review_id = f"gitea-push-{payload.repository.id}-{payload.after[:8]}-{uuid.uuid4().hex[:8]}"
+
+    # 背景執行審查 (委派給 Service)
+    service = get_gitea_webhook_service()
+    background_tasks.add_task(
+        service.review_push,
+        repo=payload.repository,
+        commits=commits,
+        sender=payload.sender,
+        review_id=review_id,
+        ref=ref,
+        before_sha=payload.before,
+        after_sha=payload.after,
+    )
+
+    logger.info(
+        "gitea_push_review_scheduled",
+        review_id=review_id,
+        repo=payload.repository.full_name,
+        ref=ref,
+        commit_count=len(commits),
+        after_sha=payload.after[:8] if payload.after else None,
+    )
+
+    return GiteaWebhookResponse(
+        status="accepted",
+        message=f"Push with {len(commits)} commit(s) review scheduled",
+        event_type="push",
+        review_id=review_id,
+    )
+
+
+# =============================================================================
+# Query Endpoints
+# =============================================================================
+
+@router.get(
+    "/reviews/{review_id}",
+    summary="取得審查結果",
+    description="根據 review_id 取得 Gitea 代碼審查結果",
+)
+async def get_review_result(review_id: str):
+    """
+    取得 Gitea 審查結果 (透過 Service)
+
+    Args:
+        review_id: 審查 ID
+
+    Returns:
+        dict: 審查結果
+    """
+    service = get_gitea_webhook_service()
+    result = await service.get_review_result(review_id)
+
+    if not result:
+        raise HTTPException(
+            status_code=status.HTTP_404_NOT_FOUND,
+            detail=f"Review {review_id} not found or expired",
+        )
+
+    return result