wait4xandGLM-5.3
938558ad4c
- T2 (op-artalk 车道)：gitea heatmap 构建期拉取+fallback+空态三保险，
53×7 灰阶格复用 tokens（亮暗自动），日界锚定 Asia/Shanghai，零 JS，
zh/en 首页挂载；fallback 经死端点实测不破构建
- T3：webhook_receiver.py（HMAC 强制+main-only+构建锁）以 systemd
--user 常驻 13021；gitea hook 注册（events=push, branch=main）
- 三个深坑：gitea 容器回宿主须走桥网关 IP 而非 127.0.0.1；gitea 28
egress 拦私网须 ALLOWED_HOST_LIST（ALLOWLIST 是废弃名）；systemd
用户服务缺用户 PATH 致 build 静默失败（stamp 不更新即症状）
验证: 测试投递 202→构建→staging 戳更新（09:48:37 实测）；
dist 热力图 376 格真数据（9 次贡献）zh/en 双页
Spec: 2026-10-25_add-blog-launch
Co-authored-by: GLM-5.3
89 lines
3.2 KiB
Python
89 lines
3.2 KiB
Python
#!/usr/bin/env python3
|
||
"""gitea webhook 接收器（Phase 4 T3）：127.0.0.1/桥网关 :13021。
|
||
|
||
安全模型：
|
||
- /hook 仅接受 HMAC-SHA256 通过的 push(main) 事件（密钥宿主 0600 文件）；
|
||
- /healthz 无副作用探活；
|
||
- 监听 0.0.0.0 是为 gitea 容器经桥网关（172.22.0.1）回连——宿主在 NAT 后、
|
||
ESA 只代理 80/443 vhost，该端口不对公网；HMAC 是硬门槛。
|
||
|
||
触发链：gitea push(main) → 本接收器验签 → build.sh → www-staging 更新。
|
||
常驻方式：systemd --user（blog-webhook.service）；reboot 自启需一次性
|
||
`sudo loginctl enable-linger wait4x`（已列入上线检查单）。
|
||
"""
|
||
import hashlib
|
||
import hmac
|
||
import json
|
||
import subprocess
|
||
import threading
|
||
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
|
||
from pathlib import Path
|
||
|
||
PORT = 13021
|
||
SECRET_FILE = Path.home() / "docker_apps/gitea/webhook-secret.txt"
|
||
BUILD = Path.home() / "Code/Github/my-blog/infra/deploy/build.sh"
|
||
BUILD_LOCK = threading.Lock()
|
||
|
||
|
||
def read_secret() -> bytes:
|
||
return SECRET_FILE.read_text().strip().encode()
|
||
|
||
|
||
class Handler(BaseHTTPRequestHandler):
|
||
def _json(self, code: int, obj: dict) -> None:
|
||
body = json.dumps(obj).encode()
|
||
self.send_response(code)
|
||
self.send_header("Content-Type", "application/json")
|
||
self.send_header("Content-Length", str(len(body)))
|
||
self.end_headers()
|
||
self.wfile.write(body)
|
||
|
||
def do_GET(self) -> None:
|
||
if self.path == "/healthz":
|
||
self._json(200, {"ok": True, "service": "blog-webhook"})
|
||
else:
|
||
self._json(404, {"error": "not found"})
|
||
|
||
def do_POST(self) -> None:
|
||
if self.path != "/hook":
|
||
self._json(404, {"error": "not found"})
|
||
return
|
||
length = int(self.headers.get("Content-Length", 0))
|
||
payload = self.rfile.read(length)
|
||
sig = self.headers.get("X-Gitea-Signature", "")
|
||
expected = hmac.new(read_secret(), payload, hashlib.sha256).hexdigest()
|
||
if not hmac.compare_digest(sig, expected):
|
||
self._json(401, {"error": "bad signature"})
|
||
return
|
||
try:
|
||
ref = json.loads(payload).get("ref", "")
|
||
except json.JSONDecodeError:
|
||
self._json(400, {"error": "bad json"})
|
||
return
|
||
if not ref.endswith("/main"):
|
||
self._json(200, {"ignored": ref})
|
||
return
|
||
if BUILD_LOCK.locked():
|
||
self._json(202, {"queued": False, "reason": "build in progress"})
|
||
return
|
||
threading.Thread(target=self._run_build, daemon=True).start()
|
||
self._json(202, {"queued": True})
|
||
|
||
def _run_build(self) -> None:
|
||
with BUILD_LOCK:
|
||
subprocess.run(
|
||
["bash", str(BUILD)],
|
||
capture_output=True, timeout=1200,
|
||
cwd=str(BUILD.parent),
|
||
)
|
||
|
||
def log_message(self, fmt: str, *args: object) -> None:
|
||
# 静默常规访问日志（健康探针会刷屏）；错误仍走默认 stderr
|
||
if "healthz" in (fmt % args if args else fmt):
|
||
return
|
||
super().log_message(fmt, *args)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
ThreadingHTTPServer(("0.0.0.0", PORT), Handler).serve_forever()
|