my-cd-webhook/bin/cd-target

#!/usr/bin/env python3
"""Query targets.json.

The manifest is the only place that knows how a (repo, branch) pair maps onto a
host, a compose stack and an image. Everything else -- the deploy script, the
hook generator, the installer -- asks this helper rather than parsing JSON
itself, so there is exactly one implementation of the matching rules.

Subcommands:
    resolve --repo OWNER/NAME --branch BRANCH [--host HOST]
        Print shell-quoted KEY=VALUE lines for the matching target.
        Exit 3 when nothing matches (a normal, non-error outcome: it just
        means this push is not ours to act on).

    list [--host HOST]
        Print the name of every enabled target owned by HOST, one per line.

    repos [--host HOST]
        Print the distinct repositories HOST deploys, one per line.

    host-config [--host HOST]
        Print shell-quoted BIND/PORT for HOST.

    hostnames
        Print the names this machine answers to (diagnostics).

HOST defaults to the local machine in every subcommand.
"""

from __future__ import annotations

import argparse
import json
import os
import shlex
import socket
import sys
from pathlib import Path

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

# CD_TARGETS_FILE lets a test run against a scratch manifest. It changes which
# manifest is read, never which host this machine is allowed to act as, so it
# cannot be used to make one host deploy another's targets.
MANIFEST = Path(os.environ.get("CD_TARGETS_FILE", REPO_ROOT / "targets.json"))

NO_MATCH = 3


def local_hostnames() -> set[str]:
    """Every name this machine plausibly answers to.

    Hosts in this fleet are referred to by FQDN in the manifest but report a
    short name from `hostname`, so both forms -- and the leading label of any
    FQDN -- have to count as a match. Getting this wrong is how the old
    per-project scripts ended up with a dead `hostname != s2` guard.
    """
    names = {socket.gethostname()}
    try:
        names.add(socket.getfqdn())
    except OSError:
        pass
    for name in list(names):
        if "." in name:
            names.add(name.split(".", 1)[0])
    return {n.lower() for n in names if n}


def host_matches(target_host: str, names: set[str]) -> bool:
    target_host = target_host.lower()
    return target_host in names or target_host.split(".", 1)[0] in names


def load() -> dict:
    try:
        return json.loads(MANIFEST.read_text())
    except FileNotFoundError:
        sys.exit(f"cd-target: manifest not found: {MANIFEST}")
    except json.JSONDecodeError as exc:
        sys.exit(f"cd-target: manifest is not valid JSON: {exc}")


def pick_host(manifest: dict, explicit: str | None) -> str:
    """Return the manifest's canonical name for the host we are acting as."""
    if explicit:
        return explicit
    names = local_hostnames()
    for host in manifest.get("hosts", {}):
        if host_matches(host, names):
            return host
    # Not a configured host: return the short local name so callers report
    # something truthful rather than silently adopting another host's identity.
    return socket.gethostname()


def emit(pairs: dict[str, object]) -> None:
    for key, value in pairs.items():
        print(f"{key}={shlex.quote(str(value))}")


def cmd_resolve(args, manifest: dict) -> int:
    host = pick_host(manifest, args.host)
    names = local_hostnames() if not args.host else {args.host.lower()}
    defaults = manifest.get("defaults", {})

    for target in manifest.get("targets", []):
        if not target.get("enabled", True):
            continue
        if target["repo"] != args.repo or target["branch"] != args.branch:
            continue
        if not host_matches(target["host"], names):
            continue

        merged = {**defaults, **target}
        merged["resolved_host"] = host
        emit({f"CD_{k.upper()}": v for k, v in merged.items() if not k.startswith("_")})
        return 0

    return NO_MATCH


def cmd_list(args, manifest: dict) -> int:
    names = local_hostnames() if not args.host else {args.host.lower()}
    for target in manifest.get("targets", []):
        if target.get("enabled", True) and host_matches(target["host"], names):
            print(target["name"])
    return 0


def cmd_repos(args, manifest: dict) -> int:
    names = local_hostnames() if not args.host else {args.host.lower()}
    seen: list[str] = []
    for target in manifest.get("targets", []):
        if not target.get("enabled", True):
            continue
        if host_matches(target["host"], names) and target["repo"] not in seen:
            seen.append(target["repo"])
    print("\n".join(seen))
    return 0


def cmd_host_config(args, manifest: dict) -> int:
    host = pick_host(manifest, args.host)
    config = manifest.get("hosts", {}).get(host)
    if config is None:
        sys.exit(
            f"cd-target: host {host!r} has no entry in targets.json 'hosts'. "
            "Add one before installing here."
        )
    emit({"CD_HOST": host, "CD_BIND": config["bind"], "CD_PORT": config["port"]})
    return 0


def cmd_hostnames(_args, _manifest: dict) -> int:
    print(" ".join(sorted(local_hostnames())))
    return 0


def main() -> int:
    parser = argparse.ArgumentParser(description="Query the CD target manifest")
    sub = parser.add_subparsers(dest="command", required=True)

    p = sub.add_parser("resolve")
    p.add_argument("--repo", required=True)
    p.add_argument("--branch", required=True)
    p.add_argument("--host")
    p.set_defaults(func=cmd_resolve)

    for name, func in (("list", cmd_list), ("repos", cmd_repos),
                       ("host-config", cmd_host_config)):
        p = sub.add_parser(name)
        p.add_argument("--host")
        p.set_defaults(func=func)

    p = sub.add_parser("hostnames")
    p.set_defaults(func=cmd_hostnames)

    args = parser.parse_args()
    return args.func(args, load())


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