webgpu-portfolio/webhook/server.py

#!/usr/bin/env python3
"""
Gitea webhook receiver — listens for push events and triggers a docker-compose
redeploy with zero-downtime (rebuild + restart only the changed service).

Usage:
    WEBHOOK_SECRET=... REPO_DIR=/home/djoser/webgpu-portfolio python3 server.py

Environment:
    WEBHOOK_SECRET  Gitea webhook secret (HMAC-SHA256)         [required]
    REPO_DIR        Path to the repo with docker-compose.yml   [default: .]
    LISTEN_HOST     Bind address                               [default: 0.0.0.0]
    LISTEN_PORT     Port                                       [default: 9000]
"""

import hashlib
import hmac
import http.server
import json
import os
import subprocess
import sys
from pathlib import Path


# ---- config ----

SECRET = os.environ.get("WEBHOOK_SECRET", "").encode()
REPO_DIR = Path(os.environ.get("REPO_DIR", Path.cwd()))
HOST = os.environ.get("LISTEN_HOST", "0.0.0.0")
PORT = int(os.environ.get("LISTEN_PORT", "9000"))


def verify_signature(body: bytes, header: str) -> bool:
    """Verify Gitea's X-Gitea-Signature header against the shared secret."""
    if not SECRET:
        print("[webhook] WARNING: WEBHOOK_SECRET is empty — accepting all requests", flush=True)
        return True
    if not header:
        return False
    expected = hmac.new(SECRET, body, hashlib.sha256).hexdigest()
    return hmac.compare_digest(expected, header)


def deploy() -> tuple[int, str]:
    """Run 'git pull' + 'docker compose up -d --build' in REPO_DIR."""
    steps = [
        # Pull latest code
        (["git", "pull", "origin", "master"], "git pull"),
        # Rebuild + restart only the web container (nginx-proxy doesn't change)
        (["docker", "compose", "up", "-d", "--build", "--remove-orphans"], "docker compose up"),
    ]

    for cmd, label in steps:
        result = subprocess.run(
            cmd,
            cwd=REPO_DIR,
            capture_output=True,
            text=True,
            timeout=120,
        )
        if result.returncode != 0:
            return 1, f"{label} failed (exit {result.returncode}):\n{result.stderr.strip()}"
        print(f"[webhook] {label}: ok", flush=True)

    return 0, "deploy ok"


class WebhookHandler(http.server.BaseHTTPRequestHandler):
    def log_message(self, fmt: str, *args) -> None:
        print(f"[webhook] {self.client_address[0]} — {fmt % args}", flush=True)

    def do_POST(self) -> None:
        content_length = int(self.headers.get("Content-Length", "0"))
        body = self.rfile.read(content_length)

        # Verify signature
        signature = self.headers.get("X-Gitea-Signature", "")
        if not verify_signature(body, signature):
            self.send_error(403, "Bad signature")
            return

        # Parse event
        event_type = self.headers.get("X-Gitea-Event", "")
        if event_type != "push":
            self.send_response(200)
            self.end_headers()
            self.wfile.write(f'{{"status":"ignored","event":"{event_type}"}}\n'.encode())
            return

        try:
            payload = json.loads(body)
        except json.JSONDecodeError:
            self.send_error(400, "Invalid JSON")
            return

        ref = payload.get("ref", "")
        print(f"[webhook] push on {ref} — deploying...", flush=True)

        exit_code, msg = deploy()
        if exit_code == 0:
            self.send_response(200)
            self.end_headers()
            self.wfile.write(b'{"status":"ok"}\n')
        else:
            self.send_response(500)
            self.end_headers()
            self.wfile.write(f'{{"status":"error","detail":{json.dumps(msg)}}}\n'.encode())

    def do_GET(self) -> None:
        """Health-check endpoint."""
        if self.path == "/health":
            self.send_response(200)
            self.end_headers()
            self.wfile.write(b"ok\n")
        else:
            self.send_error(405, "Method not allowed")


def main() -> None:
    if not SECRET:
        print(
            "[webhook] WARNING: WEBHOOK_SECRET is empty. Set it for security.\n"
            "          Generate one:  openssl rand -hex 32",
            flush=True,
        )

    server = http.server.HTTPServer((HOST, PORT), WebhookHandler)
    print(f"[webhook] Listening on {HOST}:{PORT}  (repo dir: {REPO_DIR})", flush=True)
    try:
        server.serve_forever()
    except KeyboardInterrupt:
        print("\n[webhook] Shutting down.", flush=True)
        server.shutdown()


if __name__ == "__main__":
    main()