ERP/infra/deploy-webhook/deploy_webhook.py

#!/usr/bin/env python3
"""Hardened Gitea push-to-deploy webhook receiver.

Replaces an unverified "POST /deploy" endpoint. Runs on the VPS (host port
9000 is reserved for it) and rebuilds the stack when a push to the
production branch arrives — but ONLY after verifying the Gitea webhook
signature (HMAC-SHA256 of the raw body with GITEA_WEBHOOK_SECRET).

Why this matters: an unauthenticated deploy endpoint lets anyone on the
internet restart/rebuild the production stack (and, if it shells out to a
script, is effectively remote code execution).

Setup on the VPS (see infra/deploy-webhook/README.md):

    GITEA_WEBHOOK_SECRET=... DEPLOY_DIR=/srv/erp DEPLOY_BRANCH=production \
        python3 deploy_webhook.py            # or run under systemd (provided)

Gitea side: repo → Settings → Webhooks → Add webhook → Gitea
    Target URL: http://<server>:9000/deploy
    Method: POST, Secret: same value as GITEA_WEBHOOK_SECRET
    Trigger on: Push Events, Branch filter: production

Stdlib only — no dependencies to install.
"""
import hashlib
import hmac
import json
import logging
import os
import subprocess
import sys
import time
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer

LISTEN_HOST = os.environ.get("WEBHOOK_LISTEN_HOST", "127.0.0.1")
LISTEN_PORT = int(os.environ.get("WEBHOOK_PORT", "9000"))
SECRET = os.environ.get("GITEA_WEBHOOK_SECRET", "")
DEPLOY_DIR = os.environ.get("DEPLOY_DIR", "/srv/erp")
DEPLOY_BRANCH = os.environ.get("DEPLOY_BRANCH", "production")
# Post-deploy readiness gate: defaults to the public API host from .env
# (both DEPLOY_HEALTH_URL and API_HOST live in the same EnvironmentFile).
HEALTH_URL = (
    os.environ.get("DEPLOY_HEALTH_URL")
    or (f"https://{os.environ['API_HOST']}/health/ready" if os.environ.get("API_HOST") else "")
)
LOCK_FILE = os.environ.get("DEPLOY_LOCK", "/tmp/erp-deploy.lock")

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s %(levelname)s deploy-webhook — %(message)s",
    handlers=[logging.StreamHandler(sys.stdout)],
)
log = logging.getLogger("deploy-webhook")

# Compose file pair used in production (base + overlay).
DEPLOY_CMD = os.environ.get(
    "DEPLOY_CMD",
    "docker compose -f docker-compose.yml -f docker-compose.prod.yml up -d --build",
)


def _verify_signature(body: bytes, header: str | None) -> bool:
    """Gitea sends X-Gitea-Signature: hex HMAC-SHA256 of the raw body."""
    if not SECRET:
        log.error("GITEA_WEBHOOK_SECRET is not set — refusing ALL requests")
        return False
    if not header:
        return False
    expected = hmac.new(SECRET.encode(), body, hashlib.sha256).hexdigest()
    return hmac.compare_digest(expected, header.strip().lower())


def _deploy() -> tuple[bool, str]:
    """Run the rebuild. Serialized: concurrent pushes queue/drop, not race."""
    if os.path.exists(LOCK_FILE):
        age = time.time() - os.path.getmtime(LOCK_FILE)
        if age < 1800:  # a deploy is already running (or crashed <30min ago)
            return False, f"deploy already in progress (lock age {int(age)}s)"
    try:
        with open(LOCK_FILE, "w") as f:
            f.write(str(os.getpid()))
        log.info("deploy started: %s", DEPLOY_CMD)
        proc = subprocess.run(
            DEPLOY_CMD, shell=True, cwd=DEPLOY_DIR,
            capture_output=True, text=True, timeout=1800,
        )
        tail = "\n".join((proc.stdout + proc.stderr).splitlines()[-40:])
        ok = proc.returncode == 0
        log.info("deploy finished rc=%s\n%s", proc.returncode, tail)
        if ok and HEALTH_URL:
            ok = _wait_healthy()
        return ok, ("ok" if ok else "deploy or health check failed")
    except subprocess.TimeoutExpired:
        log.error("deploy timed out after 1800s")
        return False, "deploy timed out"
    except Exception as exc:
        log.exception("deploy failed")
        return False, str(exc)
    finally:
        try:
            os.remove(LOCK_FILE)
        except OSError:
            pass


def _wait_healthy(timeout: int = 180) -> bool:
    """Poll the app's readiness endpoint after a rebuild."""
    import urllib.request
    log.info("waiting for %s", HEALTH_URL)
    deadline = time.time() + timeout
    while time.time() < deadline:
        try:
            with urllib.request.urlopen(HEALTH_URL, timeout=5) as resp:
                if resp.status == 200:
                    log.info("post-deploy health check: OK")
                    return True
        except Exception:
            pass
        time.sleep(5)
    log.error("post-deploy health check FAILED after %ss", timeout)
    return False


class Handler(BaseHTTPRequestHandler):
    def do_POST(self):
        if self.path != "/deploy":
            self._json(404, {"error": "not found"})
            return
        length = int(self.headers.get("Content-Length", "0"))
        if length > 1_000_000:
            self._json(413, {"error": "payload too large"})
            return
        body = self.rfile.read(length)
        if not _verify_signature(body, self.headers.get("X-Gitea-Signature")):
            log.warning("rejected request: bad/missing signature from %s", self.client_address[0])
            self._json(401, {"error": "invalid signature"})
            return
        try:
            payload = json.loads(body)
            ref = payload.get("ref", "")
            pushed = (payload.get("ref") or "").rsplit("/", 1)[-1]
            if ref != f"refs/heads/{DEPLOY_BRANCH}":
                log.info("ignoring push to %s (deploys trigger on %s only)", ref, DEPLOY_BRANCH)
                self._json(200, {"status": "ignored", "ref": ref})
                return
            pusher = payload.get("pusher", {}).get("login", "unknown")
            log.info("push to %s by %s — deploying", pushed, pusher)
        except Exception:
            log.exception("bad payload")
            self._json(400, {"error": "bad payload"})
            return
        # Answer the webhook quickly; deploy runs detached from the response.
        ok, msg = _deploy()
        self._json(200 if ok else 500, {"status": "ok" if ok else "failed", "detail": msg})

    def do_GET(self):
        # Minimal liveness for the receiver itself (monitoring can scrape it).
        if self.path == "/ping":
            self._json(200, {"status": "ok", "branch": DEPLOY_BRANCH})
            return
        self._json(404, {"error": "not found"})

    def _json(self, code: int, obj: dict):
        data = json.dumps(obj).encode()
        self.send_response(code)
        self.send_header("Content-Type", "application/json")
        self.send_header("Content-Length", str(len(data)))
        self.end_headers()
        self.wfile.write(data)

    def log_message(self, fmt, *args):  # route stdlib access logs to our logger
        log.debug("access: " + fmt, *args)


if __name__ == "__main__":
    if not SECRET:
        log.warning("starting WITHOUT signature verification is not supported — set GITEA_WEBHOOK_SECRET")
        sys.exit(1)
    server = ThreadingHTTPServer((LISTEN_HOST, LISTEN_PORT), Handler)
    log.info("listening on %s:%s (branch=%s dir=%s)", LISTEN_HOST, LISTEN_PORT, DEPLOY_BRANCH, DEPLOY_DIR)
    server.serve_forever()