Add deploy webhook receiver for snotes

deploy/webhook-server.py

@@ -0,0 +1,143 @@
+#!/usr/bin/env python3
+"""
+Gitea push-webhook receiver for snotes auto-deploy.
+
+Runs deploy/deploy.sh on every push event to the main branch.
+Designed to run on the Docker host.
+
+Usage:
+    python webhook-server.py --port 9997 --secret "hunter2"
+    python webhook-server.py --port 9997 --secret "hunter2" --deploy /path/to/deploy.sh
+"""
+
+from __future__ import annotations
+
+import argparse
+import hmac
+import json
+import logging
+import os
+import subprocess
+import sys
+from http.server import BaseHTTPRequestHandler, HTTPServer
+
+logging.basicConfig(
+    level=logging.INFO,
+    format="%(asctime)s deploy-webhook %(levelname)s %(message)s",
+)
+log = logging.getLogger("deploy-webhook")
+
+
+class DeployHandler(BaseHTTPRequestHandler):
+    secret: str = ""
+    deploy_script: str = ""
+
+    def log_message(self, format: str, *args: object) -> None:  # noqa: A002
+        log.info(format, *args)
+
+    def _verify_signature(self, body: bytes) -> bool:
+        if not self.secret:
+            return True
+        sig_header = self.headers.get("X-Gitea-Signature", "")
+        expected = hmac.new(
+            self.secret.encode("utf-8"), body, "sha256"
+        ).hexdigest()
+        return hmac.compare_digest(sig_header, expected)
+
+    def do_POST(self) -> None:
+        if self.path != "/deploy":
+            self.send_response(404)
+            self.end_headers()
+            return
+
+        content_len = int(self.headers.get("Content-Length", 0))
+        body = self.rfile.read(content_len)
+
+        if not self._verify_signature(body):
+            log.warning("Signature verification failed")
+            self.send_response(403)
+            self.end_headers()
+            self.wfile.write(b"Forbidden")
+            return
+
+        # Only deploy on pushes to main
+        try:
+            payload = json.loads(body)
+        except json.JSONDecodeError:
+            payload = {}
+
+        event = self.headers.get("X-Gitea-Event", "push")
+        ref = payload.get("ref", "")
+
+        if event != "push" or not ref.endswith("/main"):
+            log.info("Skipping non-push or non-main event: %s %s", event, ref)
+            self.send_response(204)
+            self.end_headers()
+            return
+
+        log.info("Push to main detected — deploying...")
+        self.send_response(202)
+        self.send_header("Content-Type", "text/plain")
+        self.end_headers()
+        self.wfile.write(b"Deploy triggered\n")
+
+        # Run deploy in background — don't block the webhook response
+        try:
+            result = subprocess.run(
+                ["bash", self.deploy_script],
+                capture_output=True,
+                text=True,
+                timeout=300,
+            )
+            for line in result.stdout.splitlines():
+                log.info("[deploy] %s", line)
+            for line in result.stderr.splitlines():
+                log.warning("[deploy] %s", line)
+            if result.returncode == 0:
+                log.info("Deploy succeeded")
+            else:
+                log.error("Deploy failed (exit %d)", result.returncode)
+        except subprocess.TimeoutExpired:
+            log.error("Deploy timed out after 300s")
+        except Exception as exc:
+            log.error("Deploy error: %s", exc)
+
+
+def main() -> None:
+    parser = argparse.ArgumentParser(description="snotes deploy webhook receiver")
+    parser.add_argument("--port", type=int, default=9997, help="Listen port")
+    parser.add_argument("--secret", default="", help="Gitea webhook secret")
+    parser.add_argument(
+        "--deploy",
+        default=os.path.join(os.path.dirname(__file__), "deploy.sh"),
+        help="Path to deploy script",
+    )
+    args = parser.parse_args()
+
+    DeployHandler.secret = args.secret
+    DeployHandler.deploy_script = os.path.abspath(args.deploy)
+
+    if not os.path.exists(DeployHandler.deploy_script):
+        log.error("Deploy script not found: %s", DeployHandler.deploy_script)
+        sys.exit(1)
+
+    if not os.access(DeployHandler.deploy_script, os.X_OK):
+        log.error("Deploy script is not executable: %s", DeployHandler.deploy_script)
+        sys.exit(1)
+
+    server = HTTPServer(("0.0.0.0", args.port), DeployHandler)
+    log.info(
+        "Listening on :%d, secret=%s, deploy=%s",
+        args.port,
+        "yes" if args.secret else "no",
+        DeployHandler.deploy_script,
+    )
+    try:
+        server.serve_forever()
+    except KeyboardInterrupt:
+        log.info("Shutting down")
+        server.server_close()
+
+
+if __name__ == "__main__":
+    main()