144 lines
4.4 KiB
Python
Executable File
144 lines
4.4 KiB
Python
Executable File
#!/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()
|