006bb24215
- deploy.sh: git pull, pip install, migrate, collectstatic, restart gunicorn - webhook_receiver.py: HTTP server con verifica HMAC-SHA256 Gitea - olimpic-nastri-webhook.service: systemd unit per il receiver - Nginx: aggiunto proxy /webhook/deploy → porta 9000 - sudoers: restart gunicorn senza password per deploy automatico
86 lines
2.3 KiB
Python
86 lines
2.3 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Webhook receiver per Gitea — Diario Conversazioni Olimpic Nastri.
|
|
Ascolta su porta 9000, verifica la firma HMAC del payload Gitea,
|
|
e lancia deploy.sh quando riceve un push sul branch main.
|
|
"""
|
|
|
|
import hashlib
|
|
import hmac
|
|
import json
|
|
import subprocess
|
|
import sys
|
|
from http.server import HTTPServer, BaseHTTPRequestHandler
|
|
|
|
WEBHOOK_SECRET = "c91dca86b9a87d9f25e07a63354da9f2469998f9"
|
|
DEPLOY_SCRIPT = "/home/marco/olimpic_nastri/deploy.sh"
|
|
LISTEN_PORT = 9000
|
|
|
|
|
|
class WebhookHandler(BaseHTTPRequestHandler):
|
|
|
|
def do_POST(self):
|
|
if self.path != "/deploy":
|
|
self.send_response(404)
|
|
self.end_headers()
|
|
return
|
|
|
|
content_length = int(self.headers.get("Content-Length", 0))
|
|
if content_length > 1_000_000: # max 1 MB payload
|
|
self.send_response(413)
|
|
self.end_headers()
|
|
return
|
|
|
|
body = self.rfile.read(content_length)
|
|
|
|
# Verifica firma HMAC-SHA256 di Gitea
|
|
signature = self.headers.get("X-Gitea-Signature", "")
|
|
expected = hmac.new(
|
|
WEBHOOK_SECRET.encode(), body, hashlib.sha256
|
|
).hexdigest()
|
|
|
|
if not hmac.compare_digest(signature, expected):
|
|
self.send_response(403)
|
|
self.end_headers()
|
|
self.wfile.write(b"Firma non valida")
|
|
return
|
|
|
|
# Controlla che sia un push su main
|
|
try:
|
|
payload = json.loads(body)
|
|
except json.JSONDecodeError:
|
|
self.send_response(400)
|
|
self.end_headers()
|
|
return
|
|
|
|
ref = payload.get("ref", "")
|
|
if ref != "refs/heads/main":
|
|
self.send_response(200)
|
|
self.end_headers()
|
|
self.wfile.write(b"Push ignorato (non main)")
|
|
return
|
|
|
|
# Lancia il deploy in background
|
|
subprocess.Popen([DEPLOY_SCRIPT], close_fds=True)
|
|
|
|
self.send_response(200)
|
|
self.end_headers()
|
|
self.wfile.write(b"Deploy avviato")
|
|
|
|
def log_message(self, format, *args):
|
|
print(f"[webhook] {args[0]}", flush=True)
|
|
|
|
|
|
def main():
|
|
server = HTTPServer(("127.0.0.1", LISTEN_PORT), WebhookHandler)
|
|
print(f"Webhook receiver in ascolto su 127.0.0.1:{LISTEN_PORT}", flush=True)
|
|
try:
|
|
server.serve_forever()
|
|
except KeyboardInterrupt:
|
|
pass
|
|
server.server_close()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|