127 lines
3.6 KiB
Python
127 lines
3.6 KiB
Python
|
#!/usr/bin/env python3
|
|||
|
|
# -*- coding: utf-8 -*-
|
|||
|
|
"""
|
|||
|
|
Gitea Webhook 处理器
|
|||
|
|
验证签名并解析 Webhook 事件
|
|||
|
|
"""
|
|||
|
|
import hmac
|
|||
|
|
import hashlib
|
|||
|
|
import logging
|
|||
|
|
from typing import Dict, Any, Optional
|
|||
|
|
|
|||
|
|
logger = logging.getLogger(__name__)
|
|||
|
|
|
|||
|
|
|
|||
|
|
class GiteaWebhookHandler:
|
|||
|
|
"""Gitea Webhook 处理器"""
|
|||
|
|
|
|||
|
|
def __init__(self, config: Dict[str, Any]):
|
|||
|
|
"""
|
|||
|
|
初始化 Webhook 处理器
|
|||
|
|
|
|||
|
|
Args:
|
|||
|
|
config: Gitea 配置
|
|||
|
|
"""
|
|||
|
|
self.config = config
|
|||
|
|
self.base_url = config.get('base_url', 'http://localhost:3000')
|
|||
|
|
self.webhook_secret = config.get('webhook_secret', '')
|
|||
|
|
|
|||
|
|
def verify_signature(self, payload: bytes, signature: str, secret: str) -> bool:
|
|||
|
|
"""
|
|||
|
|
验证 Webhook 签名
|
|||
|
|
|
|||
|
|
Args:
|
|||
|
|
payload: 请求体
|
|||
|
|
signature: 请求头中的签名
|
|||
|
|
secret: 密钥
|
|||
|
|
|
|||
|
|
Returns:
|
|||
|
|
签名是否有效
|
|||
|
|
"""
|
|||
|
|
if not secret:
|
|||
|
|
logger.warning('未配置 Webhook 密钥，跳过验证')
|
|||
|
|
return True
|
|||
|
|
|
|||
|
|
try:
|
|||
|
|
# Gitea 使用 SHA256 HMAC
|
|||
|
|
expected_signature = hmac.new(
|
|||
|
|
secret.encode('utf-8'),
|
|||
|
|
payload,
|
|||
|
|
hashlib.sha256
|
|||
|
|
).hexdigest()
|
|||
|
|
|
|||
|
|
# 比较签名（使用 constant time 比较防止时序攻击）
|
|||
|
|
return hmac.compare_digest(f'sha256={expected_signature}', signature)
|
|||
|
|
except Exception as e:
|
|||
|
|
logger.error(f'签名验证失败: {str(e)}')
|
|||
|
|
return False
|
|||
|
|
|
|||
|
|
def parse_push_event(self, payload: Dict[str, Any]) -> Dict[str, Any]:
|
|||
|
|
"""
|
|||
|
|
解析 Push 事件
|
|||
|
|
|
|||
|
|
Args:
|
|||
|
|
payload: Webhook payload
|
|||
|
|
|
|||
|
|
Returns:
|
|||
|
|
解析后的提交信息
|
|||
|
|
"""
|
|||
|
|
repo = payload.get('repository', {})
|
|||
|
|
commits = payload.get('commits', [])
|
|||
|
|
ref = payload.get('ref', '')
|
|||
|
|
|
|||
|
|
return {
|
|||
|
|
'repo_name': repo.get('full_name', ''),
|
|||
|
|
'repo_url': repo.get('clone_url', ''),
|
|||
|
|
'web_url': repo.get('web_url', ''),
|
|||
|
|
'branch': ref.replace('refs/heads/', ''),
|
|||
|
|
'commits': [
|
|||
|
|
{
|
|||
|
|
'id': commit.get('id', '')[:8],
|
|||
|
|
'message': commit.get('message', ''),
|
|||
|
|
'author': commit.get('author', {}).get('name', ''),
|
|||
|
|
'email': commit.get('author', {}).get('email', ''),
|
|||
|
|
'timestamp': commit.get('timestamp', ''),
|
|||
|
|
'added': commit.get('added', []),
|
|||
|
|
'modified': commit.get('modified', []),
|
|||
|
|
'removed': commit.get('removed', []),
|
|||
|
|
}
|
|||
|
|
for commit in commits
|
|||
|
|
],
|
|||
|
|
'pusher': payload.get('pusher', {}).get('name', ''),
|
|||
|
|
'before': payload.get('before', ''),
|
|||
|
|
'after': payload.get('after', ''),
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
def get_changed_files(self, commit: Dict[str, Any]) -> list:
|
|||
|
|
"""
|
|||
|
|
获取提交中变更的文件列表
|
|||
|
|
|
|||
|
|
Args:
|
|||
|
|
commit: 提交信息
|
|||
|
|
|
|||
|
|
Returns:
|
|||
|
|
变更的文件列表
|
|||
|
|
"""
|
|||
|
|
files = []
|
|||
|
|
files.extend(commit.get('added', []))
|
|||
|
|
files.extend(commit.get('modified', []))
|
|||
|
|
files.extend(commit.get('removed', []))
|
|||
|
|
return files
|
|||
|
|
|
|||
|
|
def filter_by_extension(self, files: list, extensions: list) -> list:
|
|||
|
|
"""
|
|||
|
|
按文件扩展名过滤文件
|
|||
|
|
|
|||
|
|
Args:
|
|||
|
|
files: 文件列表
|
|||
|
|
extensions: 扩展名列表（如 ['.py', '.js']）
|
|||
|
|
|
|||
|
|
Returns:
|
|||
|
|
过滤后的文件列表
|
|||
|
|
"""
|
|||
|
|
return [
|
|||
|
|
f for f in files
|
|||
|
|
if any(f.endswith(ext) for ext in extensions)
|
|||
|
|
]
|