92 lines
3.4 KiB
Python
92 lines
3.4 KiB
Python
|
#!/usr/bin/env python3
|
||
|
|
# Simple webhook server for Gitea
|
||
|
|
# Listens for push events and triggers auto-deploy
|
||
|
|
|
||
|
|
import http.server
|
||
|
|
import socketserver
|
||
|
|
import json
|
||
|
|
import subprocess
|
||
|
|
import os
|
||
|
|
import hmac
|
||
|
|
import hashlib
|
||
|
|
from urllib.parse import urlparse
|
||
|
|
|
||
|
|
# Configuration
|
||
|
|
PORT = 8080
|
||
|
|
SECRET = 'your-webhook-secret' # Change this!
|
||
|
|
DEPLOY_SCRIPT = '/opt/scripts/auto-deploy.sh'
|
||
|
|
|
||
|
|
class WebhookHandler(http.server.BaseHTTPRequestHandler):
|
||
|
|
def do_POST(self):
|
||
|
|
if self.path == '/webhook':
|
||
|
|
try:
|
||
|
|
# Read the payload
|
||
|
|
content_length = int(self.headers['Content-Length'])
|
||
|
|
post_data = self.rfile.read(content_length)
|
||
|
|
|
||
|
|
# Verify signature (optional but recommended)
|
||
|
|
signature = self.headers.get('X-Gitea-Signature', '')
|
||
|
|
if signature:
|
||
|
|
expected_sig = hmac.new(
|
||
|
|
SECRET.encode(),
|
||
|
|
post_data,
|
||
|
|
hashlib.sha256
|
||
|
|
).hexdigest()
|
||
|
|
expected_sig = f'sha256={expected_sig}'
|
||
|
|
|
||
|
|
if not hmac.compare_digest(signature, expected_sig):
|
||
|
|
self.send_response(401)
|
||
|
|
self.end_headers()
|
||
|
|
self.wfile.write(b'Invalid signature')
|
||
|
|
return
|
||
|
|
|
||
|
|
# Parse the webhook payload
|
||
|
|
data = json.loads(post_data.decode('utf-8'))
|
||
|
|
|
||
|
|
# Only deploy on manual trigger (with "deploy" in commit message)
|
||
|
|
commit_message = data.get('commits', [{}])[0].get('message', '').lower()
|
||
|
|
|
||
|
|
if 'deploy now' in commit_message or 'manual deploy' in commit_message:
|
||
|
|
print(f"Manual deploy requested: {commit_message}")
|
||
|
|
|
||
|
|
# Run the deploy script in background
|
||
|
|
subprocess.Popen([
|
||
|
|
'nohup', 'bash', '/opt/scripts/deploy-now.sh'
|
||
|
|
], stdout=open('/var/log/webhook-deploy.log', 'a'),
|
||
|
|
stderr=subprocess.STDOUT)
|
||
|
|
|
||
|
|
self.send_response(200)
|
||
|
|
self.end_headers()
|
||
|
|
self.wfile.write(b'Manual deployment triggered')
|
||
|
|
else:
|
||
|
|
print(f"Ignoring push - not a manual deploy trigger")
|
||
|
|
self.send_response(200)
|
||
|
|
self.end_headers()
|
||
|
|
self.wfile.write(b'Push received - daily deploy scheduled for 3AM')
|
||
|
|
|
||
|
|
except Exception as e:
|
||
|
|
print(f"Webhook error: {e}")
|
||
|
|
self.send_response(500)
|
||
|
|
self.end_headers()
|
||
|
|
self.wfile.write(b'Internal server error')
|
||
|
|
else:
|
||
|
|
self.send_response(404)
|
||
|
|
self.end_headers()
|
||
|
|
self.wfile.write(b'Not found')
|
||
|
|
|
||
|
|
def do_GET(self):
|
||
|
|
if self.path == '/health':
|
||
|
|
self.send_response(200)
|
||
|
|
self.end_headers()
|
||
|
|
self.wfile.write(b'OK')
|
||
|
|
else:
|
||
|
|
self.send_response(404)
|
||
|
|
self.end_headers()
|
||
|
|
self.wfile.write(b'Not found')
|
||
|
|
|
||
|
|
if __name__ == '__main__':
|
||
|
|
os.makedirs('/var/log', exist_ok=True)
|
||
|
|
|
||
|
|
with socketserver.TCPServer(("", PORT), WebhookHandler) as httpd:
|
||
|
|
print(f"Webhook server listening on port {PORT}")
|
||
|
|
httpd.serve_forever()
|