77 lines
3.1 KiB
Python
Executable File
77 lines
3.1 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""Simple webhook listener for Gitea push events to trigger supersam deploy."""
|
|
import json
|
|
import hashlib
|
|
import hmac
|
|
import subprocess
|
|
import logging
|
|
from http.server import HTTPServer, BaseHTTPRequestHandler
|
|
|
|
SECRET = '1032ef91aacd726907bb72c023813c6827f56354902cc7b30e72626690c6d51d'
|
|
|
|
logging.basicConfig(level=logging.INFO, format='%(asctime)s %(levelname)s %(message)s')
|
|
logger = logging.getLogger('webhook')
|
|
|
|
class WebhookHandler(BaseHTTPRequestHandler):
|
|
def verify_signature(self, body):
|
|
signature = self.headers.get('X-Gitea-Signature', '')
|
|
if not signature:
|
|
return False
|
|
computed = hmac.new(SECRET.encode(), body, hashlib.sha256).hexdigest()
|
|
return hmac.compare_digest(signature, computed)
|
|
|
|
def do_POST(self):
|
|
if self.path == '/webhook/supersam':
|
|
content_length = int(self.headers.get('Content-Length', 0))
|
|
body = self.rfile.read(content_length) if content_length else b''
|
|
if not self.verify_signature(body):
|
|
self.send_response(403)
|
|
self.end_headers()
|
|
self.wfile.write(b'{"error": "invalid signature"}')
|
|
return
|
|
try:
|
|
payload = json.loads(body) if body else {}
|
|
ref = payload.get('ref', '')
|
|
if ref == 'refs/heads/main':
|
|
logger.info('Push to main detected, starting deploy...')
|
|
result = subprocess.run(
|
|
['/opt/supersam/deploy.sh'],
|
|
capture_output=True, text=True, timeout=600
|
|
)
|
|
logger.info('Deploy exit code: %s', result.returncode)
|
|
if result.returncode != 0:
|
|
logger.error('Deploy stderr: %s', result.stderr[-2000:])
|
|
self.send_response(200)
|
|
self.send_header('Content-Type', 'application/json')
|
|
self.end_headers()
|
|
self.wfile.write(json.dumps({'status': 'deployed', 'exit_code': result.returncode}).encode())
|
|
else:
|
|
logger.info('Ignoring push to %s', ref)
|
|
self.send_response(200)
|
|
self.send_header('Content-Type', 'application/json')
|
|
self.end_headers()
|
|
self.wfile.write(b'{"status": "ignored"}')
|
|
except Exception as e:
|
|
logger.error('Error: %s', e)
|
|
self.send_response(500)
|
|
self.end_headers()
|
|
self.wfile.write(json.dumps({'error': str(e)}).encode())
|
|
else:
|
|
self.send_response(404)
|
|
self.end_headers()
|
|
|
|
def do_GET(self):
|
|
if self.path == '/health':
|
|
self.send_response(200)
|
|
self.send_header('Content-Type', 'application/json')
|
|
self.end_headers()
|
|
self.wfile.write(b'{"status": "ok"}')
|
|
else:
|
|
self.send_response(404)
|
|
self.end_headers()
|
|
|
|
if __name__ == '__main__':
|
|
server = HTTPServer(('0.0.0.0', 9000), WebhookHandler)
|
|
logger.info('Webhook listener on http://0.0.0.0:9000')
|
|
server.serve_forever()
|