my-cd-webhook/bin/cd-render-hooks

#!/usr/bin/env python3
"""Render an adnanh/webhook hooks file for this host from targets.json.

One hook is emitted per repository that this host actually deploys, not one per
environment. That keeps the Gitea side trivial -- a repository gets exactly one
webhook, configured once -- while the branch/environment mapping stays in
targets.json where it can change without anyone touching Gitea.

Secrets never live in this repository. They are read from a host-local file
(default ~/.config/cd-webhook/secrets.env, mode 0600) as:

    CD_WEBHOOK_SECRET_WEBDEV_DOMAINDINGO=<the secret set in Gitea>

A repository with no secret is a hard error: a hook without HMAC validation
would accept a deploy request from anyone who can reach the port.

Usage:
    cd-render-hooks [--host HOST] [--output PATH] [--redact]
    cd-render-hooks --list-urls [--output PATH] [--base-url URL]
"""

from __future__ import annotations

import argparse
import importlib.machinery
import importlib.util
import json
import os
import re
import stat
import sys
from pathlib import Path

BIN_DIR = Path(__file__).resolve().parent
REPO_ROOT = BIN_DIR.parent


def _load_cd_target():
    """Import bin/cd-target, whose hyphenated name blocks a plain import.

    Sharing the module rather than re-implementing the lookup keeps one
    definition of how a (repo, branch, host) triple is matched.
    """
    spec = importlib.util.spec_from_loader(
        "cd_target",
        importlib.machinery.SourceFileLoader("cd_target", str(BIN_DIR / "cd-target")),
    )
    module = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(module)
    return module


_cd_target = _load_cd_target()
load = _cd_target.load
pick_host = _cd_target.pick_host
names_for = _cd_target.names_for
enabled_targets = _cd_target.enabled_targets

DEFAULT_CONFIG_DIR = Path(
    os.environ.get("XDG_CONFIG_HOME", Path.home() / ".config")
) / "cd-webhook"
DEFAULT_SECRETS = DEFAULT_CONFIG_DIR / "secrets.env"
DEFAULT_OUTPUT = DEFAULT_CONFIG_DIR / "hooks.json"

REDACTED = "<redacted>"


def secret_var(repo: str) -> str:
    """webdev/domaindingo -> CD_WEBHOOK_SECRET_WEBDEV_DOMAINDINGO"""
    return "CD_WEBHOOK_SECRET_" + re.sub(r"[^A-Za-z0-9]+", "_", repo).upper()


def hook_id(repo: str) -> str:
    """webdev/domaindingo -> webdev-domaindingo (becomes the URL path)."""
    return re.sub(r"[^A-Za-z0-9]+", "-", repo).lower().strip("-")


def load_secrets(path: Path) -> dict[str, str]:
    if not path.exists():
        sys.exit(
            f"cd-render-hooks: no secrets file at {path}\n"
            f"  Create it (mode 0600) with one line per repository, e.g.\n"
            f"    CD_WEBHOOK_SECRET_WEBDEV_DOMAINDINGO=<secret from the Gitea webhook>"
        )

    mode = path.stat().st_mode
    if mode & (stat.S_IRWXG | stat.S_IRWXO):
        sys.exit(
            f"cd-render-hooks: {path} is group/world accessible.\n"
            f"  Run: chmod 600 {path}"
        )

    secrets: dict[str, str] = {}
    for raw in path.read_text().splitlines():
        line = raw.strip()
        if not line or line.startswith("#") or "=" not in line:
            continue
        key, _, value = line.partition("=")
        secrets[key.strip()] = value.strip().strip('"').strip("'")
    return secrets


def build_hook(repo: str, branches: list[str], secret: str) -> dict:
    """One hook: HMAC-validated, push-only, restricted to known branches."""
    branch_rules = [
        {
            "match": {
                "type": "value",
                "value": f"refs/heads/{branch}",
                "parameter": {"source": "payload", "name": "ref"},
            }
        }
        for branch in branches
    ]

    return {
        "id": hook_id(repo),
        "execute-command": str(REPO_ROOT / "bin" / "cd-deploy"),
        "command-working-directory": str(REPO_ROOT),
        "http-methods": ["POST"],
        "incoming-payload-content-type": "application/json",

        # The deploy runs detached; Gitea gets an immediate answer and never
        # times out waiting for an image build plus a health check.
        "response-message": f"deploy request accepted for {repo}",
        "include-command-output-in-response": False,

        # A push to a branch we do not deploy is a normal event, not a failure.
        # Returning 200 keeps Gitea's delivery history readable; the daemon log
        # records which rule did not match.
        "trigger-rule-mismatch-http-response-code": 200,

        "pass-arguments-to-command": [
            {"source": "string", "name": "--repo"},
            {"source": "payload", "name": "repository.full_name"},
            {"source": "string", "name": "--ref"},
            {"source": "payload", "name": "ref"},
            {"source": "string", "name": "--sha"},
            {"source": "payload", "name": "after"},
            {"source": "string", "name": "--event"},
            {"source": "header", "name": "X-Gitea-Event"},
        ],

        "trigger-rule": {
            "and": [
                {
                    "match": {
                        "type": "payload-hmac-sha256",
                        "secret": secret,
                        "parameter": {
                            "source": "header",
                            "name": "X-Gitea-Signature",
                        },
                    }
                },
                {
                    "match": {
                        "type": "value",
                        "value": "push",
                        "parameter": {
                            "source": "header",
                            "name": "X-Gitea-Event",
                        },
                    }
                },
                {"or": branch_rules} if len(branch_rules) > 1 else branch_rules[0],
            ]
        },
    }


def list_urls(path: Path, base_url: str) -> int:
    """Print one endpoint URL per hook in an already-rendered hooks file.

    Reads the rendered file rather than the manifest so that what is printed is
    what the daemon is actually serving. Both the installer and cd-status use
    this instead of reaching into the file's schema themselves.
    """
    try:
        hooks = json.loads(path.read_text())
    except FileNotFoundError:
        sys.exit(f"cd-render-hooks: no rendered hooks file at {path}")
    for hook in hooks:
        print(f"{base_url}/hooks/{hook['id']}")
    return 0


def main() -> int:
    parser = argparse.ArgumentParser(description=__doc__.splitlines()[0])
    parser.add_argument("--host")
    parser.add_argument("--secrets", type=Path, default=DEFAULT_SECRETS)
    parser.add_argument("--output", type=Path, default=DEFAULT_OUTPUT)
    parser.add_argument(
        "--redact",
        action="store_true",
        help="print to stdout with secrets masked, and write nothing",
    )
    parser.add_argument(
        "--list-urls",
        action="store_true",
        help="print the endpoint URL of each hook in an existing hooks file",
    )
    parser.add_argument(
        "--base-url",
        default="",
        help="prefix for --list-urls, e.g. http://10.255.255.1:21600",
    )
    args = parser.parse_args()

    if args.list_urls:
        return list_urls(args.output, args.base_url.rstrip("/"))

    manifest = load()
    host = pick_host(manifest, args.host)

    # repo -> branches, preserving manifest order for a stable diff.
    by_repo: dict[str, list[str]] = {}
    for target in enabled_targets(manifest, names_for(host)):
        by_repo.setdefault(target["repo"], []).append(target["branch"])

    if not by_repo:
        sys.exit(
            f"cd-render-hooks: no enabled targets for host {host!r}.\n"
            f"  Either this host deploys nothing, or its name does not match "
            f"any 'host' field in targets.json."
        )

    secrets = {} if args.redact else load_secrets(args.secrets)

    hooks = []
    for repo, branches in by_repo.items():
        if args.redact:
            secret = REDACTED
        else:
            var = secret_var(repo)
            secret = secrets.get(var, "")
            if not secret:
                sys.exit(
                    f"cd-render-hooks: {args.secrets} has no value for {var}\n"
                    f"  Every deployed repository needs its Gitea webhook secret here."
                )
        hooks.append(build_hook(repo, branches, secret))

    rendered = json.dumps(hooks, indent=2) + "\n"

    if args.redact:
        print(rendered, end="")
        return 0

    args.output.parent.mkdir(parents=True, exist_ok=True)
    # Create with restrictive permissions before any secret reaches the disk.
    fd = os.open(args.output, os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600)
    with os.fdopen(fd, "w") as handle:
        handle.write(rendered)

    print(f"wrote {args.output} ({len(hooks)} hook(s) for {host})")
    for repo, branches in by_repo.items():
        print(f"  /hooks/{hook_id(repo)}  <-  {repo}  [{', '.join(branches)}]")
    return 0


if __name__ == "__main__":
    sys.exit(main())