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 result: 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.

    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 functools
import json
import os
import shlex
import socket
import sys
from collections.abc import Iterator
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


@functools.cache
def local_hostnames() -> frozenset[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.

    Cached: `getfqdn` can block on a slow resolver, and this sits on the path of
    every webhook delivery. The answer cannot change within one process.
    """
    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 frozenset(n.lower() for n in names if n)


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


def names_for(explicit: str | None) -> frozenset[str]:
    """The names a target's `host` field is matched against.

    Defaults to this machine's own names. An explicit --host is expanded to both
    its FQDN and short forms, the same way `local_hostnames` expands the local
    ones, so `--host s5` and `--host s5.fisher.hu` select the same targets.
    """
    if not explicit:
        return local_hostnames()
    name = explicit.lower()
    return frozenset({name, name.split(".", 1)[0]})


def enabled_targets(manifest: dict, names: frozenset[str]) -> Iterator[dict]:
    """Every enabled target owned by `names`, in manifest order.

    The single implementation of "a host acts only on its own targets" -- the
    rule the whole design rests on. cd-render-hooks imports this rather than
    repeating the filter, so a new filtering dimension is one edit, not four.
    """
    for target in manifest.get("targets", []):
        if target.get("enabled", True) and host_matches(target["host"], names):
            yield target


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 shell_value(value: object) -> str:
    """Render a manifest value the way a shell consumer expects it.

    JSON booleans would otherwise arrive as Python's `True`/`False`, forcing
    every consumer to test for both spellings of every boolean field.
    """
    if isinstance(value, bool):
        return "true" if value else "false"
    return str(value)


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


def cmd_resolve(args, manifest: dict) -> int:
    host = pick_host(manifest, args.host)
    defaults = manifest.get("defaults", {})

    for target in enabled_targets(manifest, names_for(args.host)):
        if target["repo"] != args.repo or target["branch"] != args.branch:
            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:
    for target in enabled_targets(manifest, names_for(args.host)):
        print(target["name"])
    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), ("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())