chore: disable old K3s Actions; add webhook deploy scripts

2026-09-23 15:16:40 +08:00
parent b1ef04ee30
commit a6a3d88624
4 changed files with 143 additions and 0 deletions
@@ -0,0 +1,76 @@
+#!/usr/bin/env python3
+"""Gitea push webhook → only refs/heads/dev → scripts/deploy-dev.sh"""
+from __future__ import annotations
+
+import hashlib
+import hmac
+import json
+import os
+import subprocess
+import threading
+from http.server import BaseHTTPRequestHandler, HTTPServer
+
+SECRET = os.environ.get("WEBHOOK_SECRET", "").encode()
+DEPLOY_SH = os.environ.get("DEPLOY_SH", "/opt/AirShelf/scripts/deploy-dev.sh")
+HOST = os.environ.get("WEBHOOK_HOST", "0.0.0.0")
+PORT = int(os.environ.get("WEBHOOK_PORT", "9000"))
+
+
+def verify(sig: str | None, body: bytes) -> bool:
+    if not SECRET or not sig:
+        return False
+    digest = hmac.new(SECRET, body, hashlib.sha256).hexdigest()
+    return hmac.compare_digest(digest, sig.strip())
+
+
+def run_deploy() -> None:
+    subprocess.run([DEPLOY_SH], check=False)
+
+
+class Handler(BaseHTTPRequestHandler):
+    def log_message(self, fmt: str, *args) -> None:
+        print(f"[webhook] {self.address_string()} - {fmt % args}")
+
+    def do_POST(self) -> None:
+        if self.path not in ("/hooks/deploy-dev", "/deploy-dev", "/"):
+            self.send_response(404)
+            self.end_headers()
+            return
+        length = int(self.headers.get("Content-Length", "0"))
+        body = self.rfile.read(length)
+        sig = self.headers.get("X-Gitea-Signature") or self.headers.get("X-Hub-Signature-256", "")
+        if sig.startswith("sha256="):
+            sig = sig[7:]
+        if not verify(sig, body):
+            self.send_response(401)
+            self.end_headers()
+            self.wfile.write(b"invalid signature")
+            return
+        try:
+            payload = json.loads(body.decode("utf-8") or "{}")
+        except json.JSONDecodeError:
+            self.send_response(400)
+            self.end_headers()
+            return
+        ref = payload.get("ref", "")
+        if ref != "refs/heads/dev":
+            self.send_response(200)
+            self.end_headers()
+            self.wfile.write(f"ignored ref={ref}".encode())
+            return
+        threading.Thread(target=run_deploy, daemon=True).start()
+        self.send_response(200)
+        self.end_headers()
+        self.wfile.write(b"deploy-dev accepted")
+
+    def do_GET(self) -> None:
+        self.send_response(200)
+        self.end_headers()
+        self.wfile.write(b"airshelf webhook ok")
+
+
+if __name__ == "__main__":
+    if not SECRET:
+        raise SystemExit("WEBHOOK_SECRET is required")
+    print(f"listening on {HOST}:{PORT}, deploy={DEPLOY_SH}")
+    HTTPServer((HOST, PORT), Handler).serve_forever()