diff --git a/apps/api/src/api/v1/gitea_webhook.py b/apps/api/src/api/v1/gitea_webhook.py new file mode 100644 index 00000000..39892647 --- /dev/null +++ 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= + - 使用 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