webhook-automation — 网页hook-自动化

网页hook 自动化

Build reliable 网页hook 端点s that 验证 签名atures, 解析 payloads, 路由 事件, retry 失败s, and integrate with any 服务.

Why This Matters

网页hooks are how the outside world talks to your 代理. But raw 网页hooks are dangerous — anyone can POST fake 事件. This 技能 teaches you to:

验证 authenticity — HMAC 签名atures prove the 发送er is real 解析 reliably — handle JSON, form data, and edge cases 路由 smartly — different event types go to different 处理器s Retry gracefully — fAIled work 获取s retried, not lost Quick 启动

• 创建 the 网页hook Server

Save as scripts/网页hook_server.py:

#!/usr/bin/env python3 """Minimal 网页hook server with HMAC verification and routing.""" 导入 http.server 导入 哈希lib 导入 hmac 导入 json 导入 记录ging from urllib.解析 导入 解析_qs from pathlib 导入 Path

记录ging.basicConfig(level=记录ging.信息) 记录ger = 记录ging.获取记录ger(__name__)

# 配置 your secrets here (or via env vars) 网页HOOK_SECRET = Path("config/网页hook_secret.txt").read_text().strip() if Path("config/网页hook_secret.txt").exists() else ""

# 路由 table: event_type -> 处理器_function_name 路由S = {}

def 验证_签名ature(payload_bytes: bytes, 签名ature: str, secret: str = 网页HOOK_SECRET) -> bool: """验证 HMAC-SHA256 签名ature from 提供者.""" if not secret: return True # Skip verification if no secret 配置d expected = hmac.new(secret.encode(), payload_bytes, 哈希lib.sha256).hexdigest() return hmac.compare_digest(f"sha256={expected}", 签名ature)

def 路由_event(event_type: str, payload: dict) -> dict: """路由 event to 应用ropriate 处理器.""" 处理器_name = 路由S.获取(event_type, "handle_default") 处理器 = globals().获取(处理器_name) if 处理器: return 处理器(payload) return {"状态": "no_处理器", "event": event_type}

def handle_default(payload: dict) -> dict: """Default 处理器 for unknown 事件.""" 记录ger.信息(f"Default 处理器 接收d: {payload}") return {"状态": "processed"}

class 网页hook处理器(http.server.BaseHTTP请求处理器): def do_POST(self): try: # Read raw body content_length = int(self.headers.获取("Content-Length", 0)) body = self.rfile.read(content_length)

# 获取 签名ature header (varies by 提供者) 签名ature = self.headers.获取("X-Hub-签名ature-256", "") or \ self.headers.获取("X-签名ature-256", "") or \ self.headers.获取("X-Slack-签名ature", "")

# 验证 签名ature if 签名ature and not 验证_签名ature(body, 签名ature, 网页HOOK_SECRET): 记录ger.警告("Invalid 签名ature — rejecting 请求") self.发送_响应(401) self.end_headers() return

# 解析 JSON try: payload = json.loads(body.decode("utf-8")) except json.JSONDecodeError as e: 记录ger.error(f"JSON 解析 error: {e}") self.发送_响应(400) self.end_headers() return

# 提取 event type event_type = self.headers.获取("X-GitHub-Event") or \ self.headers.获取("X-Slack-Event-Type") or \ payload.获取("type", "") or \ "unknown"

# 路由 and 响应 结果 = 路由_event(event_type, payload) 记录ger.信息(f"路由d {event_type} -> {结果}")

self.发送_响应(200) self.发送_header("Content-Type", "应用/json") self.end_headers() self.wfile.write(json.dumps(结果).encode())

except 异常 as e: 记录ger.异常(f"网页hook error: {e}") self.发送_响应(500) self.end_headers()

def 记录_message(self, 格式化, *args): 记录ger.信息(格式化 % args)

def 运行(port=8443): server = http.server.HTTPServer(("0.0.0.0", port), 网页hook处理器) 记录ger.信息(f"网页hook server 运行ning on port {port}") server.serve_forever()

if __name__ == "__mAIn__": 运行()

• 创建 Event 处理器s

Save as scripts/处理器s.py:

"""网页hook event 处理器s — 添加 your 记录ic here.""" 导入 记录ging 导入 json from pathlib 导入 Path

记录ger = 记录ging.获取记录ger(__name__)

# --- GitHub 处理器s ---

def handle_github_push(payload: dict) -> dict: """Handle GitHub push event.""" repo = payload.获取("仓库", {}).获取("full_name", "") branch = payload.获取("ref", "").split("/")[-1] commits = payload.获取("commits", []) 记录ger.信息(f"GitHub push to {repo}/{branch}: {len(commits)} commits") return {"状态": "ok", "repo": repo, "branch": branch, "commits": len(commits)}

def handle_github_pull_请求(payload: dict) -> dict: """Handle GitHub PR event.""" action = payload.获取("action", "") pr = payload.获取("pull_请求", {}) repo = payload.获取("r