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