← Retour
#!/usr/bin/env python3
"""
GitLab Webhook Receiver - Auto-deploy on push
33800 Stack CI/CD
Usage:
python3 webhook-receiver.py --env dev --port 5500
python3 webhook-receiver.py --env prod --port 5500
Environment:
WEBHOOK_SECRET: GitLab webhook secret token
"""
import http.server
import json
import os
import subprocess
import sys
import argparse
import hashlib
import hmac
from datetime import datetime
# Configuration
APPS_DIR = "/home/gouroubleu/apps"
LOG_FILE = "/var/log/webhook-deploy.log"
NOTIFY_URL = "http://192.168.1.12:5300/api/notify/push"
def log(message):
"""Log message to file and stdout"""
timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
line = f"[{timestamp}] {message}"
print(line)
try:
with open(LOG_FILE, "a") as f:
f.write(line + "\n")
except:
pass
def notify(title, body, level="info"):
"""Send notification via notif-logger"""
try:
import urllib.request
data = json.dumps({"title": title, "body": body, "level": level}).encode()
req = urllib.request.Request(NOTIFY_URL, data=data, headers={"Content-Type": "application/json"})
urllib.request.urlopen(req, timeout=5)
except Exception as e:
log(f"Notification failed: {e}")
def verify_token(request_token, expected_token):
"""Verify GitLab webhook token"""
if not expected_token:
return True # No token configured = accept all
return hmac.compare_digest(request_token or "", expected_token)
def get_config_file(project_dir, env):
"""Get the conf.{env}.gouroubleu.yml file path"""
config_file = os.path.join(project_dir, f"conf.{env}.gouroubleu.yml")
if os.path.exists(config_file):
return config_file
return None
def deploy_project(project_name, branch, env):
"""Deploy a project"""
project_dir = os.path.join(APPS_DIR, project_name)
if not os.path.exists(project_dir):
log(f"Project directory not found: {project_dir}")
return False, "Project not found"
# Check if config file exists
config_file = get_config_file(project_dir, env)
if not config_file:
log(f"No config file conf.{env}.gouroubleu.yml found for {project_name}")
return False, "No config file"
log(f"Deploying {project_name} ({branch}) to {env}...")
try:
# Git pull
log(f" Git fetch and checkout {branch}...")
subprocess.run(
["git", "fetch", "origin", branch],
cwd=project_dir, check=True, capture_output=True
)
subprocess.run(
["git", "checkout", branch],
cwd=project_dir, check=True, capture_output=True
)
subprocess.run(
["git", "pull", "origin", branch],
cwd=project_dir, check=True, capture_output=True
)
# Docker compose build and up (--no-cache to ensure code changes are rebuilt)
log(f" Building containers (no-cache)...")
result = subprocess.run(
["docker", "compose", "build", "--no-cache"],
cwd=project_dir, capture_output=True, text=True
)
if result.returncode != 0:
log(f" Build failed: {result.stderr}")
return False, "Build failed"
result = subprocess.run(
["docker", "compose", "up", "-d"],
cwd=project_dir, capture_output=True, text=True
)
if result.returncode != 0:
log(f" Deploy failed: {result.stderr}")
return False, "Deploy failed"
log(f" Deploy successful!")
return True, "Success"
except subprocess.CalledProcessError as e:
log(f" Command failed: {e}")
return False, str(e)
except Exception as e:
log(f" Error: {e}")
return False, str(e)
class WebhookHandler(http.server.BaseHTTPRequestHandler):
env = "dev"
secret = None
def log_message(self, format, *args):
"""Override to use our logging"""
log(f"HTTP: {args[0]}")
def do_GET(self):
"""Health check endpoint"""
if self.path == "/health":
self.send_response(200)
self.send_header("Content-Type", "application/json")
self.end_headers()
self.wfile.write(json.dumps({"status": "ok", "env": self.env}).encode())
else:
self.send_response(404)
self.end_headers()
def do_POST(self):
"""Handle webhook"""
if self.path != "/webhook":
self.send_response(404)
self.end_headers()
return
# Verify token
request_token = self.headers.get("X-Gitlab-Token")
if not verify_token(request_token, self.secret):
log("Invalid webhook token")
self.send_response(401)
self.end_headers()
return
# Read body
content_length = int(self.headers.get("Content-Length", 0))
body = self.rfile.read(content_length)
try:
payload = json.loads(body)
except json.JSONDecodeError:
log("Invalid JSON payload")
self.send_response(400)
self.end_headers()
return
# Extract info from GitLab push event
event_type = self.headers.get("X-Gitlab-Event", "")
if event_type != "Push Hook":
log(f"Ignoring event type: {event_type}")
self.send_response(200)
self.send_header("Content-Type", "application/json")
self.end_headers()
self.wfile.write(json.dumps({"status": "ignored", "reason": "not a push event"}).encode())
return
# Get project and branch
project_name = payload.get("project", {}).get("name", "").lower()
ref = payload.get("ref", "") # refs/heads/develop
branch = ref.replace("refs/heads/", "")
log(f"Received push: {project_name} ({branch})")
# Check if branch matches environment
expected_branch = "develop" if self.env == "dev" else "main"
if branch != expected_branch:
log(f"Ignoring branch {branch} (expected {expected_branch} for {self.env})")
self.send_response(200)
self.send_header("Content-Type", "application/json")
self.end_headers()
self.wfile.write(json.dumps({
"status": "ignored",
"reason": f"branch {branch} not configured for {self.env}"
}).encode())
return
# Deploy
success, message = deploy_project(project_name, branch, self.env)
# Notify
if success:
notify(
f"Deploy {project_name}",
f"Branch {branch} deploye sur {self.env.upper()}",
"info"
)
else:
notify(
f"Deploy FAILED {project_name}",
f"Branch {branch} sur {self.env.upper()}: {message}",
"warning"
)
# Response
self.send_response(200 if success else 500)
self.send_header("Content-Type", "application/json")
self.end_headers()
self.wfile.write(json.dumps({
"status": "success" if success else "error",
"project": project_name,
"branch": branch,
"env": self.env,
"message": message
}).encode())
def main():
parser = argparse.ArgumentParser(description="GitLab Webhook Receiver")
parser.add_argument("--env", choices=["dev", "prod"], required=True, help="Environment")
parser.add_argument("--port", type=int, default=5500, help="Port to listen on")
parser.add_argument("--secret", help="Webhook secret token (or WEBHOOK_SECRET env var)")
args = parser.parse_args()
# Get secret from args or env
secret = args.secret or os.environ.get("WEBHOOK_SECRET")
# Configure handler
WebhookHandler.env = args.env
WebhookHandler.secret = secret
# Start server
server = http.server.HTTPServer(("0.0.0.0", args.port), WebhookHandler)
log(f"Webhook receiver started on port {args.port} for {args.env.upper()}")
log(f"Endpoints: GET /health, POST /webhook")
if secret:
log(f"Token verification: enabled")
else:
log(f"Token verification: disabled (no secret configured)")
try:
server.serve_forever()
except KeyboardInterrupt:
log("Shutting down...")
server.shutdown()
if __name__ == "__main__":
main()